ob_root_service.h 57.0 KB
Newer Older
O
oceanbase-admin 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131
/**
 * Copyright (c) 2021 OceanBase
 * OceanBase CE is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *          http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */

#ifndef OCEANBASE_ROOTSERVER_OB_ROOT_SERVICE_H_
#define OCEANBASE_ROOTSERVER_OB_ROOT_SERVICE_H_

#include "lib/thread_local/thread_buffer.h"
#include "lib/net/ob_addr.h"
#include "lib/thread/ob_work_queue.h"
#include "ob_log_archive_scheduler.h"

#include "share/backup/ob_backup_info_mgr.h"
#include "share/ob_common_rpc_proxy.h"
#include "share/ob_freeze_info_proxy.h"
#include "share/ob_inner_config_root_addr.h"
#include "share/ob_root_addr_agent.h"
#include "share/ob_web_service_root_addr.h"
#include "share/ob_remote_sql_proxy.h"
#include "share/ob_unit_replica_counter.h"
#include "share/partition_table/ob_inmemory_partition_table.h"

#include "sql/ob_index_sstable_builder.h"

#include "rpc/ob_packet.h"
#include "observer/ob_restore_ctx.h"
#include "rootserver/restore/ob_restore_scheduler.h"
#include "rootserver/restore/ob_restore_info.h"
#include "rootserver/ob_server_manager.h"
#include "rootserver/ob_all_server_task.h"
#include "rootserver/ob_all_server_checker.h"
#include "rootserver/ob_ddl_service.h"
#include "rootserver/ob_zone_manager.h"
#include "rootserver/ob_root_major_freeze_v2.h"
#include "rootserver/ob_root_minor_freeze.h"
#include "rootserver/ob_unit_manager.h"
#include "rootserver/ob_daily_merge_scheduler.h"
#include "rootserver/ob_partition_table_util.h"
#include "rootserver/ob_leader_coordinator.h"
#include "rootserver/ob_major_freeze_launcher.h"
#include "rootserver/ob_freeze_info_updater.h"
#include "rootserver/ob_global_max_decided_trans_version_mgr.h"
#include "rootserver/ob_vtable_location_getter.h"
#include "rootserver/ob_root_balancer.h"
#include "rootserver/ob_rebalance_task_mgr.h"
#include "rootserver/ob_rebalance_task.h"
#include "rootserver/ob_system_admin_util.h"
#include "rootserver/ob_rebalance_task_executor.h"
#include "rootserver/ob_root_inspection.h"
#include "rootserver/ob_lost_replica_checker.h"
#include "rootserver/ob_rs_event_history_table_operator.h"
#include "rootserver/ob_empty_server_checker.h"
#include "rootserver/ob_rs_thread_checker.h"
#include "rootserver/ob_freeze_info_manager.h"
#include "rootserver/ob_inner_table_monitor.h"
#include "rootserver/ob_global_index_builder.h"
#include "rootserver/ob_partition_spliter.h"
#include "rootserver/ob_rs_gts_manager.h"
#include "rootserver/ob_rs_gts_task_mgr.h"
#include "rootserver/ob_rs_gts_monitor.h"
#include "rootserver/ob_upgrade_storage_format_version_executor.h"
#include "rootserver/ob_schema_split_executor.h"
#include "rootserver/ob_upgrade_executor.h"
#include "rootserver/ob_upgrade_storage_format_version_executor.h"
#include "rootserver/ob_create_inner_schema_executor.h"
#include "rootserver/ob_schema_revise_executor.h"
#include "rootserver/ob_root_backup.h"
#include "rootserver/backup/ob_root_validate.h"
#include "rootserver/ob_update_rs_list_task.h"
#include "rootserver/ob_schema_history_recycler.h"
#include "rootserver/ob_backup_data_clean.h"
#include "rootserver/backup/ob_backup_auto_delete_expired_backup.h"
#include "rootserver/ob_single_partition_balance.h"
#include "rootserver/backup/ob_backup_lease_service.h"
#include "rootserver/ob_restore_point_service.h"

namespace oceanbase {

namespace common {
class ObPacket;
class ObServerConfig;
class ObConfigManager;
class ObMySQLProxy;
class ObRequestTZInfoResult;
class ObRequestTZInfoArg;
class ObPartitionKey;
class ObString;
}  // namespace common

namespace share {

namespace status {
enum ObRootServiceStatus;
}
class ObAutoincrementService;
namespace schema {
class ObMultiVersionSchemaService;
class ObTenantSchema;
class ObDatabaseSchema;
class ObTablegroupSchema;
class ObTableSchema;
class ObSchemaGetterGuard;
}  // namespace schema
class ObPartitionTableOperator;
class ObPartitionInfo;
class ObPartitionReplica;
class ObIPartitionLocationCache;
class ObRsMgr;
class ObRemotePartitionTableOperator;
}  // namespace share

namespace obrpc {
class ObSrvRpcProxy;
class ObCommonRpcProxy;
struct ObGetSwitchoverStatusRes;
}  // namespace obrpc
namespace storage {
class ObMajorFreeze;
}

namespace rootserver {
class ObSinglePartBalance;
class ObRsStatus {
G
gm 已提交
132
public:
O
oceanbase-admin 已提交
133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148
  ObRsStatus() : rs_status_(share::status::INIT)
  {}
  virtual ~ObRsStatus()
  {}
  int set_rs_status(const share::status::ObRootServiceStatus status);
  share::status::ObRootServiceStatus get_rs_status() const;
  bool need_do_restart() const;
  bool can_start_service() const;
  bool is_start() const;
  bool is_stopping() const;
  bool is_full_service() const;
  bool in_service() const;
  bool is_need_stop() const;
  int revoke_rs();
  int try_set_stopping();

G
gm 已提交
149
private:
O
oceanbase-admin 已提交
150 151 152 153 154
  common::SpinRWLock lock_;
  share::status::ObRootServiceStatus rs_status_;
};
// Root Service Entry Class
class ObRootService {
G
gm 已提交
155
public:
O
oceanbase-admin 已提交
156 157 158 159 160
  friend class TestRootServiceCreateTable_check_rs_capacity_Test;
  friend class ObStandbyIneffectSchemaTask;
  friend class ObTenantWrsTask;

  class ObStartStopServerTask : public share::ObAsyncTask {
G
gm 已提交
161
  public:
O
oceanbase-admin 已提交
162 163 164 165 166 167 168 169 170
    ObStartStopServerTask(ObRootService& root_service, const common::ObAddr& server, const bool start)
        : root_service_(root_service), server_(server), start_(start)
    {}
    virtual ~ObStartStopServerTask()
    {}
    virtual int process();
    virtual int64_t get_deep_copy_size() const;
    share::ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const;

G
gm 已提交
171
  private:
O
oceanbase-admin 已提交
172 173 174 175 176 177
    ObRootService& root_service_;
    const common::ObAddr server_;
    const bool start_;
  };

  class ObOfflineServerTask : public share::ObAsyncTask {
G
gm 已提交
178
  public:
O
oceanbase-admin 已提交
179 180 181 182 183 184 185 186 187 188 189
    ObOfflineServerTask(ObRootService& root_service, const common::ObAddr& server)
        : root_service_(root_service), server_(server)
    {
      set_retry_times(0); /*not repeat*/
    }
    virtual ~ObOfflineServerTask()
    {}
    virtual int process();
    virtual int64_t get_deep_copy_size() const;
    share::ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const;

G
gm 已提交
190
  private:
O
oceanbase-admin 已提交
191 192 193 194 195
    ObRootService& root_service_;
    const common::ObAddr server_;
  };

  class ObMergeErrorTask : public share::ObAsyncTask {
G
gm 已提交
196
  public:
O
oceanbase-admin 已提交
197 198 199 200 201 202 203 204
    explicit ObMergeErrorTask(ObRootService& root_service) : root_service_(root_service)
    {}
    virtual ~ObMergeErrorTask()
    {}
    virtual int process();
    virtual int64_t get_deep_copy_size() const;
    share::ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const;

G
gm 已提交
205
  private:
O
oceanbase-admin 已提交
206 207 208 209
    ObRootService& root_service_;
  };

  class ObStatisticPrimaryZoneCountTask : public share::ObAsyncTask {
G
gm 已提交
210
  public:
O
oceanbase-admin 已提交
211 212 213 214 215 216 217 218
    explicit ObStatisticPrimaryZoneCountTask(ObRootService& root_service) : root_service_(root_service)
    {}
    virtual ~ObStatisticPrimaryZoneCountTask()
    {}
    virtual int process();
    virtual int64_t get_deep_copy_size() const;
    share::ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const;

G
gm 已提交
219
  private:
O
oceanbase-admin 已提交
220 221 222 223
    ObRootService& root_service_;
  };

  class ObCreateHaGtsUtilTask : public share::ObAsyncTask {
G
gm 已提交
224
  public:
O
oceanbase-admin 已提交
225 226 227 228 229 230 231 232
    explicit ObCreateHaGtsUtilTask(ObRootService& root_service) : root_service_(root_service)
    {}
    virtual ~ObCreateHaGtsUtilTask()
    {}
    virtual int process();
    virtual int64_t get_deep_copy_size() const;
    share::ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const;

G
gm 已提交
233
  private:
O
oceanbase-admin 已提交
234 235 236 237
    ObRootService& root_service_;
  };

  class ObRefreshServerTask : public common::ObAsyncTimerTask {
G
gm 已提交
238
  public:
O
oceanbase-admin 已提交
239 240 241 242 243 244 245 246 247 248 249
    const static int64_t REFRESH_SERVER_INTERVAL = 1L * 1000 * 1000;  // 1 second
    explicit ObRefreshServerTask(ObRootService& root_service);
    virtual ~ObRefreshServerTask(){};
    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
250
  private:
O
oceanbase-admin 已提交
251 252 253 254 255
    ObRootService& root_service_;
    DISALLOW_COPY_AND_ASSIGN(ObRefreshServerTask);
  };

  class ObStatusChangeCallback : public ObIStatusChangeCallback {
G
gm 已提交
256
  public:
O
oceanbase-admin 已提交
257 258 259 260 261 262 263 264 265 266
    explicit ObStatusChangeCallback(ObRootService& root_service);
    virtual ~ObStatusChangeCallback();

    virtual int wakeup_balancer() override;
    virtual int wakeup_daily_merger() override;
    virtual int on_server_status_change(const common::ObAddr& server) override;
    virtual int on_start_server(const common::ObAddr& server) override;
    virtual int on_stop_server(const common::ObAddr& server) override;
    virtual int on_offline_server(const common::ObAddr& server) override;

G
gm 已提交
267
  private:
O
oceanbase-admin 已提交
268 269
    ObRootService& root_service_;

G
gm 已提交
270
  private:
O
oceanbase-admin 已提交
271 272 273 274
    DISALLOW_COPY_AND_ASSIGN(ObStatusChangeCallback);
  };

  class ObServerChangeCallback : public ObIServerChangeCallback {
G
gm 已提交
275
  public:
O
oceanbase-admin 已提交
276 277 278 279 280 281 282
    explicit ObServerChangeCallback(ObRootService& root_service) : root_service_(root_service)
    {}
    virtual ~ObServerChangeCallback()
    {}

    virtual int on_server_change() override;

G
gm 已提交
283
  private:
O
oceanbase-admin 已提交
284 285
    ObRootService& root_service_;

G
gm 已提交
286
  private:
O
oceanbase-admin 已提交
287 288 289 290
    DISALLOW_COPY_AND_ASSIGN(ObServerChangeCallback);
  };

  class ObRestartTask : public common::ObAsyncTimerTask {
G
gm 已提交
291
  public:
O
oceanbase-admin 已提交
292 293 294 295 296 297 298 299 300 301 302
    explicit ObRestartTask(ObRootService& root_service);
    virtual ~ObRestartTask();

    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
303
  private:
O
oceanbase-admin 已提交
304 305
    ObRootService& root_service_;

G
gm 已提交
306
  private:
O
oceanbase-admin 已提交
307 308 309 310
    DISALLOW_COPY_AND_ASSIGN(ObRestartTask);
  };

  class ObReportCoreTableReplicaTask : public common::ObAsyncTimerTask {
G
gm 已提交
311
  public:
O
oceanbase-admin 已提交
312 313 314 315
    explicit ObReportCoreTableReplicaTask(ObRootService& root_service);
    virtual ~ObReportCoreTableReplicaTask()
    {}

G
gm 已提交
316
  public:
O
oceanbase-admin 已提交
317 318 319 320 321 322 323 324
    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
325
  private:
O
oceanbase-admin 已提交
326 327 328 329 330
    ObRootService& root_service_;
    DISALLOW_COPY_AND_ASSIGN(ObReportCoreTableReplicaTask);
  };

  class ObReloadUnitManagerTask : public common::ObAsyncTimerTask {
G
gm 已提交
331
  public:
O
oceanbase-admin 已提交
332 333 334 335
    explicit ObReloadUnitManagerTask(ObRootService& root_service, ObUnitManager& unit_manager);
    virtual ~ObReloadUnitManagerTask()
    {}

G
gm 已提交
336
  public:
O
oceanbase-admin 已提交
337 338 339 340 341 342 343 344
    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
345
  private:
O
oceanbase-admin 已提交
346 347 348 349 350 351
    ObRootService& root_service_;
    ObUnitManager& unit_manager_;
    DISALLOW_COPY_AND_ASSIGN(ObReloadUnitManagerTask);
  };

  class ObLoadIndexBuildTask : public common::ObAsyncTimerTask {
G
gm 已提交
352
  public:
O
oceanbase-admin 已提交
353 354 355 356 357 358 359 360 361
    ObLoadIndexBuildTask(ObRootService& root_service, ObGlobalIndexBuilder& global_index_builder);
    virtual ~ObLoadIndexBuildTask() = default;
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
362
  private:
O
oceanbase-admin 已提交
363 364 365 366 367
    ObRootService& root_service_;
    ObGlobalIndexBuilder& global_index_builder_;
  };

  class ObUpdateAllServerConfigTask : public common::ObAsyncTimerTask {
G
gm 已提交
368
  public:
O
oceanbase-admin 已提交
369 370 371 372
    explicit ObUpdateAllServerConfigTask(ObRootService& root_service);
    virtual ~ObUpdateAllServerConfigTask()
    {}

G
gm 已提交
373
  public:
O
oceanbase-admin 已提交
374 375 376 377 378 379 380 381
    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
382
  private:
O
oceanbase-admin 已提交
383 384 385 386 387
    ObRootService& root_service_;
    DISALLOW_COPY_AND_ASSIGN(ObUpdateAllServerConfigTask);
  };

  class ObSelfCheckTask : public common::ObAsyncTimerTask {
G
gm 已提交
388
  public:
O
oceanbase-admin 已提交
389 390 391 392 393 394 395 396 397 398
    explicit ObSelfCheckTask(ObRootService& root_service);
    virtual ~ObSelfCheckTask(){};
    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
399
  private:
O
oceanbase-admin 已提交
400 401
    void print_error_log();

G
gm 已提交
402
  private:
O
oceanbase-admin 已提交
403 404 405 406 407
    ObRootService& root_service_;
    DISALLOW_COPY_AND_ASSIGN(ObSelfCheckTask);
  };

  class ObIndexSSTableBuildTask : public share::ObAsyncTask {
G
gm 已提交
408
  public:
O
oceanbase-admin 已提交
409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441
    ObIndexSSTableBuildTask(const sql::ObIndexSSTableBuilder::BuildIndexJob& job,
        sql::ObIndexSSTableBuilder::ReplicaPicker& replica_picker, ObGlobalIndexBuilder& global_index_builder,
        common::ObMySQLProxy& sql_proxy, common::ObOracleSqlProxy& oracle_sql_proxy, const int64_t abs_timeout_us)
        : job_(job),
          replica_picker_(replica_picker),
          global_index_builder_(global_index_builder),
          sql_proxy_(sql_proxy),
          oracle_sql_proxy_(oracle_sql_proxy),
          abs_timeout_us_(abs_timeout_us)
    {
      set_retry_times(0);
    }

    virtual int process() override;

    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }

    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

    sql::ObIndexSSTableBuilder::BuildIndexJob job_;
    sql::ObIndexSSTableBuilder::ReplicaPicker& replica_picker_;
    ObGlobalIndexBuilder& global_index_builder_;
    common::ObMySQLProxy& sql_proxy_;
    common::ObOracleSqlProxy& oracle_sql_proxy_;
    int64_t abs_timeout_us_;

    DISALLOW_COPY_AND_ASSIGN(ObIndexSSTableBuildTask);
  };

  class RsListChangeCb : public share::ObIRsListChangeCb {
G
gm 已提交
442
  public:
O
oceanbase-admin 已提交
443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460
    explicit RsListChangeCb(ObRootService& rs) : rs_(rs)
    {}
    virtual ~RsListChangeCb()
    {}
    // interface of ObIRsListChangeCb
    virtual int submit_update_rslist_task(const bool force_update = false) override
    {
      return rs_.submit_update_rslist_task(force_update);
    }
    virtual int submit_report_replica() override
    {
      return rs_.report_replica();
    }
    virtual int submit_report_replica(const common::ObPartitionKey& key) override
    {
      return rs_.report_single_replica(key);
    }

G
gm 已提交
461
  private:
O
oceanbase-admin 已提交
462 463 464 465 466
    ObRootService& rs_;
    DISALLOW_COPY_AND_ASSIGN(RsListChangeCb);
  };

  class MergeErrorCallback : public share::ObIMergeErrorCb {
G
gm 已提交
467
  public:
O
oceanbase-admin 已提交
468 469 470 471 472 473 474 475 476
    MergeErrorCallback(ObRootService& rs) : rs_(rs)
    {}
    virtual ~MergeErrorCallback()
    {}
    virtual int submit_merge_error_task() override
    {
      return rs_.submit_merge_error_task();
    }

G
gm 已提交
477
  private:
O
oceanbase-admin 已提交
478 479 480 481 482 483
    // data members
    ObRootService& rs_;
    DISALLOW_COPY_AND_ASSIGN(MergeErrorCallback);
  };

  class ObInnerTableMonitorTask : public common::ObAsyncTimerTask {
G
gm 已提交
484
  public:
O
oceanbase-admin 已提交
485 486 487 488 489 490 491 492 493 494 495 496 497
    const static int64_t PURGE_INTERVAL = 3600L * 1000L * 1000L;  // 1h
    ObInnerTableMonitorTask(ObRootService& rs);
    virtual ~ObInnerTableMonitorTask()
    {}

    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
498
  private:
O
oceanbase-admin 已提交
499 500 501 502
    ObRootService& rs_;
    DISALLOW_COPY_AND_ASSIGN(ObInnerTableMonitorTask);
  };
  class ObSwitchOverToFollowerTask : public share::ObAsyncTask {
G
gm 已提交
503
  public:
O
oceanbase-admin 已提交
504 505 506 507 508 509 510 511 512 513 514
    const static int64_t RETRY_INTERVAL = 1 * 1000L * 1000L;  // 1s
    explicit ObSwitchOverToFollowerTask(ObRootService& rs);
    virtual ~ObSwitchOverToFollowerTask()
    {}
    virtual int process();
    virtual int64_t get_deep_copy_size() const
    {
      return sizeof(*this);
    }
    virtual share::ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const;

G
gm 已提交
515
  private:
O
oceanbase-admin 已提交
516 517 518 519 520
    ObRootService& rs_;
    DISALLOW_COPY_AND_ASSIGN(ObSwitchOverToFollowerTask);
  };

  class ObPrimaryClusterInspectionTask : public share::ObAsyncTask {
G
gm 已提交
521
  public:
O
oceanbase-admin 已提交
522 523 524 525 526 527 528 529 530 531 532
    const static int64_t RETRY_INTERVAL = 100 * 1000L * 1000L;  // 100s
    explicit ObPrimaryClusterInspectionTask(ObRootService& rs);
    virtual ~ObPrimaryClusterInspectionTask()
    {}
    virtual int process();
    virtual int64_t get_deep_copy_size() const
    {
      return sizeof(*this);
    }
    virtual share::ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const;

G
gm 已提交
533
  private:
O
oceanbase-admin 已提交
534 535 536 537 538
    ObRootService& rs_;
    DISALLOW_COPY_AND_ASSIGN(ObPrimaryClusterInspectionTask);
  };

  class ObGenNextSchemaVersionTask : public common::ObAsyncTimerTask {
G
gm 已提交
539
  public:
O
oceanbase-admin 已提交
540 541 542 543 544 545 546 547 548 549 550 551 552
    const static int64_t RETRY_INTERVAL = 2 * 1000L * 1000L;  // 2s
    ObGenNextSchemaVersionTask(ObRootService& rs);
    virtual ~ObGenNextSchemaVersionTask()
    {}

    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
553
  private:
O
oceanbase-admin 已提交
554 555 556 557 558
    ObRootService& rs_;
    DISALLOW_COPY_AND_ASSIGN(ObGenNextSchemaVersionTask);
  };

  class ObLostReplicaCheckerTask : public share::ObAsyncTask {
G
gm 已提交
559
  public:
O
oceanbase-admin 已提交
560 561 562 563 564 565 566 567 568 569 570 571 572 573
    const static int64_t RETRY_INTERVAL = 3 * 1000L * 1000L;  // 10s
    ObLostReplicaCheckerTask(ObRootService& rs) : rs_(rs)
    {}
    virtual ~ObLostReplicaCheckerTask()
    {}

    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
574
  private:
O
oceanbase-admin 已提交
575 576 577 578 579
    ObRootService& rs_;
    DISALLOW_COPY_AND_ASSIGN(ObLostReplicaCheckerTask);
  };

  class ObMinorFreezeTask : public share::ObAsyncTask {
G
gm 已提交
580
  public:
O
oceanbase-admin 已提交
581 582 583 584 585 586 587 588
    explicit ObMinorFreezeTask(const obrpc::ObRootMinorFreezeArg& arg) : arg_(arg)
    {}
    virtual ~ObMinorFreezeTask()
    {}
    virtual int process();
    virtual int64_t get_deep_copy_size() const;
    share::ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const;

G
gm 已提交
589
  private:
O
oceanbase-admin 已提交
590 591 592 593
    obrpc::ObRootMinorFreezeArg arg_;
  };

  class ObUpdateClusterInfoTask : public common::ObAsyncTimerTask {
G
gm 已提交
594
  public:
O
oceanbase-admin 已提交
595 596 597 598 599 600 601 602 603 604 605
    const static int64_t REFRESH_SERVER_INTERVAL = 100L * 1000 * 1000;  // 100 second
    explicit ObUpdateClusterInfoTask(ObRootService& root_service);
    virtual ~ObUpdateClusterInfoTask(){};
    // interface of AsyncTask
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
606
  private:
O
oceanbase-admin 已提交
607 608 609 610 611
    ObRootService& root_service_;
    DISALLOW_COPY_AND_ASSIGN(ObUpdateClusterInfoTask);
  };

  class ObReloadUnitReplicaCounterTask : public common::ObAsyncTimerTask {
G
gm 已提交
612
  public:
O
oceanbase-admin 已提交
613 614 615 616 617 618 619 620 621 622
    const static int64_t REFRESH_UNIT_INTERVAL = 60 * 10L * 1000 * 1000;  // 1 second
    explicit ObReloadUnitReplicaCounterTask(ObRootService& root_service);
    virtual ~ObReloadUnitReplicaCounterTask(){};
    virtual int process() override;
    virtual int64_t get_deep_copy_size() const override
    {
      return sizeof(*this);
    }
    virtual ObAsyncTask* deep_copy(char* buf, const int64_t buf_size) const override;

G
gm 已提交
623
  private:
O
oceanbase-admin 已提交
624 625 626 627
    ObRootService& root_service_;
    DISALLOW_COPY_AND_ASSIGN(ObReloadUnitReplicaCounterTask);
  };

G
gm 已提交
628
public:
O
oceanbase-admin 已提交
629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226
  ObRootService();
  virtual ~ObRootService();
  void reset_fail_count();
  void update_fail_count(int ret);
  int fake_init(common::ObServerConfig& config, common::ObConfigManager& config_mgr, obrpc::ObSrvRpcProxy& rpc_proxy,
      obrpc::ObCommonRpcProxy& common_proxy, common::ObAddr& self, common::ObMySQLProxy& sql_proxy,
      share::ObRsMgr& rs_mgr, share::schema::ObMultiVersionSchemaService* schema_mgr_,
      share::ObPartitionTableOperator& pt_operator_, share::ObRemotePartitionTableOperator& remote_pt_operator_,
      share::ObIPartitionLocationCache& location_cache);

  int init(common::ObServerConfig& config, common::ObConfigManager& config_mgr, obrpc::ObSrvRpcProxy& rpc_proxy,
      obrpc::ObCommonRpcProxy& common_proxy, common::ObAddr& self, common::ObMySQLProxy& sql_proxy,

      share::ObRemoteSqlProxy& remote_sql_proxy, observer::ObRestoreCtx& restore_ctx, share::ObRsMgr& rs_mgr,
      share::schema::ObMultiVersionSchemaService* schema_mgr_, share::ObPartitionTableOperator& pt_operator_,
      share::ObRemotePartitionTableOperator& remote_pt_operator_, share::ObIPartitionLocationCache& location_cache);
  inline bool is_inited() const
  {
    return inited_;
  }
  void destroy();

  // add virtual make the following functions mockable
  virtual int start_service();
  int revoke_rs();
  bool is_need_stop() const;
  virtual int stop_service();
  virtual int stop();
  virtual void wait();
  virtual bool in_service() const;
  bool need_do_restart() const;
  int set_rs_status(const share::status::ObRootServiceStatus status);
  virtual bool is_full_service() const;
  virtual bool is_major_freeze_done() const
  {
    return is_full_service();
  }
  virtual bool is_ddl_allowed() const
  {
    return is_full_service();
  }
  bool can_start_service() const;
  bool is_stopping() const;
  bool is_start() const;
  share::status::ObRootServiceStatus get_status() const;
  bool in_debug() const
  {
    return debug_;
  }
  void set_debug()
  {
    debug_ = true;
  }
  int reload_config();
  virtual bool check_config(const ObConfigItem& item, const char*& err_info);
  // misc get functions
  share::ObPartitionTableOperator& get_pt_operator()
  {
    return *pt_operator_;
  }
  share::ObRemotePartitionTableOperator& get_remote_pt_operator()
  {
    return *remote_pt_operator_;
  }
  share::schema::ObMultiVersionSchemaService& get_schema_service()
  {
    return *schema_service_;
  }
  ObServerManager& get_server_mgr()
  {
    return server_manager_;
  }
  ObZoneManager& get_zone_mgr()
  {
    return zone_manager_;
  }
  ObUnitManager& get_unit_mgr()
  {
    return unit_manager_;
  }
  ObRebalanceTaskMgr& get_rebalance_task_mgr()
  {
    return rebalance_task_mgr_;
  }
  ObRootBalancer& get_root_balancer()
  {
    return root_balancer_;
  }
  ObDailyMergeScheduler& get_daily_merge_scheduler()
  {
    return daily_merge_scheduler_;
  }
  ObRestoreScheduler& get_restore_scheduler()
  {
    return restore_scheduler_;
  }
  ObRootInspection& get_root_inspection()
  {
    return root_inspection_;
  }
  ObLeaderCoordinator& get_leader_coordinator()
  {
    return leader_coordinator_;
  }
  common::ObMySQLProxy& get_sql_proxy()
  {
    return sql_proxy_;
  }
  share::ObRemoteSqlProxy& get_remote_sql_proxy()
  {
    return *remote_sql_proxy_;
  }
  obrpc::ObCommonRpcProxy& get_common_rpc_proxy()
  {
    return common_proxy_;
  }
  obrpc::ObSrvRpcProxy& get_rpc_proxy()
  {
    return rpc_proxy_;
  }
  common::ObWorkQueue& get_task_queue()
  {
    return task_queue_;
  }
  common::ObWorkQueue& get_inspect_task_queue()
  {
    return inspect_task_queue_;
  }
  common::ObServerConfig* get_server_config()
  {
    return config_;
  }
  ObGlobalIndexBuilder& get_global_index_builder()
  {
    return global_index_builder_;
  }
  int64_t get_core_meta_table_version()
  {
    return core_meta_table_version_;
  }
  ObSchemaHistoryRecycler& get_schema_history_recycler()
  {
    return schema_history_recycler_;
  }
  int admin_clear_balance_task(const obrpc::ObAdminClearBalanceTaskArg& arg);
  int check_has_enough_member(const common::ObIArray<common::ObAddr>& server_list, bool& has_enough_member);
  int check_has_need_offline_replica(
      obrpc::ObTenantSchemaVersions& tenant_schema_versions, bool& has_need_offline_replica);
  int check_server_have_enough_resource_for_delete_server(const ObIArray<ObAddr>& servers, const ObZone& zone);

  // not implemented rpc, helper function for rs rpc processor define.
  int not_implement();

  int execute_bootstrap(const obrpc::ObBootstrapArg& arg);
  int set_disable_ps_config();
  int set_max_trx_size_config();
  // one phase commit
  int set_1pc_config();
  int set_balance_strategy_config();
  int check_config_result(const char* name, const char* value);
  int set_enable_oracle_priv_check();

  int check_ddl_allowed();
  int create_sys_index_table();

  int renew_lease(const share::ObLeaseRequest& lease_request, share::ObLeaseResponse& lease_response);
  int get_root_partition(share::ObPartitionInfo& partition_info);
  int report_root_partition(const share::ObPartitionReplica& replica);
  int remove_root_partition(const common::ObAddr& server);
  int rebuild_root_partition(const common::ObAddr& server);
  int clear_rebuild_root_partition(const common::ObAddr& server);
  int fetch_location(const obrpc::UInt64& table_id, common::ObSArray<share::ObPartitionLocation>& locations);
  int merge_error(const obrpc::ObMergeErrorArg& arg);
  int handle_merge_error();
  int merge_finish(const obrpc::ObMergeFinishArg& arg);

  int try_block_server(int rc, const common::ObAddr& server);

  int statistic_primary_zone_entity_count();

  int upgrade_cluster_create_ha_gts_util();

  // balance over
  template <typename TASK, typename T>
  int do_receive_balance_over(TASK& task, const T& arg);

  int receive_balance_over(const obrpc::ObMigrateReplicaRes& res)
  {
    ObMigrateReplicaTask task;
    return do_receive_balance_over(task, res);
  }
  int receive_balance_over(const obrpc::ObChangeReplicaRes& res)
  {
    ObTypeTransformTask task;
    return do_receive_balance_over(task, res);
  }
  int receive_balance_over(const obrpc::ObAddReplicaRes& res)
  {
    ObAddReplicaTask task;
    return do_receive_balance_over(task, res);
  }
  int receive_balance_over(const obrpc::ObRebuildReplicaRes& res)
  {
    ObRebuildReplicaTask task;
    return do_receive_balance_over(task, res);
  }
  int receive_balance_over(const obrpc::ObRestoreReplicaRes& res)
  {
    ObRestoreTask task;
    return do_receive_balance_over(task, res);
  }
  int receive_balance_over(const obrpc::ObPhyRestoreReplicaRes& res)
  {
    ObPhysicalRestoreTask task;
    return do_receive_balance_over(task, res);
  }
  int receive_balance_over(const obrpc::ObValidateRes& res)
  {
    ObValidateTask task;
    root_validate_.wakeup();
    return do_receive_balance_over(task, res);
  }
  // batch balance over
  template <typename TASK, typename T>
  int do_receive_batch_balance_over(TASK& task, const T& arg);

  int receive_batch_balance_over(const obrpc::ObMigrateReplicaBatchRes& res)
  {
    ObMigrateReplicaTask task;
    return do_receive_batch_balance_over(task, res);
  }
  int receive_batch_balance_over(const obrpc::ObChangeReplicaBatchRes& res)
  {
    ObTypeTransformTask task;
    return do_receive_batch_balance_over(task, res);
  }
  int receive_batch_balance_over(const obrpc::ObAddReplicaBatchRes& res)
  {
    ObAddReplicaTask task;
    return do_receive_batch_balance_over(task, res);
  }
  int receive_batch_balance_over(const obrpc::ObRebuildReplicaBatchRes& res)
  {
    ObRebuildReplicaTask task;
    return do_receive_batch_balance_over(task, res);
  }
  int receive_batch_balance_over(const obrpc::ObStandbyCutDataBatchTaskRes& res)
  {
    UNUSED(res);
    return OB_NOT_SUPPORTED;
  }
  int receive_batch_balance_over(const obrpc::ObCopySSTableBatchRes& res)
  {
    ObCopySSTableTask task;
    return do_receive_batch_balance_over(task, res);
  }
  int receive_batch_balance_over(const obrpc::ObBackupBatchRes& res)
  {
    ObBackupTask task;
    bool has_failed_task = false;
    for (int64_t i = 0; i < res.res_array_.count() && !has_failed_task; ++i) {
      const obrpc::ObBackupRes& backup_res = res.res_array_.at(i);
      if (OB_SUCCESS != backup_res.result_) {
        has_failed_task = true;
      }
    }
    if (!has_failed_task) {
      root_backup_.wakeup();
    }
    return do_receive_batch_balance_over(task, res);
  }
  int receive_batch_balance_over(const obrpc::ObValidateBatchRes& res)
  {
    ObValidateTask task;
    root_validate_.wakeup();
    return do_receive_batch_balance_over(task, res);
  }

  int observer_copy_local_index_sstable(const obrpc::ObServerCopyLocalIndexSSTableArg& arg);

  int schedule_sql_bkgd_task(const sql::ObSchedBKGDDistTask& bkgd_task);
  int sql_bkgd_task_execute_over(const sql::ObBKGDTaskCompleteArg& arg);

  // int add_replica_over(const obrpc::ObAddReplicaRes &arg);
  // int migrate_replica_over(const obrpc::ObMigrateReplicaRes &arg);
  int admin_rebuild_replica(const obrpc::ObAdminRebuildReplicaArg& arg);
  int broadcast_ds_action(const obrpc::ObDebugSyncActionArg& arg);
  int sync_pt_finish(const obrpc::ObSyncPartitionTableFinishArg& arg);
  int sync_pg_pt_finish(const obrpc::ObSyncPGPartitionMTFinishArg& arg);
  int check_dangling_replica_finish(const obrpc::ObCheckDanglingReplicaFinishArg& arg);
  int fetch_alive_server(const obrpc::ObFetchAliveServerArg& arg, obrpc::ObFetchAliveServerResult& result);
  int get_tenant_schema_versions(
      const obrpc::ObGetSchemaArg& arg, obrpc::ObTenantSchemaVersions& tenant_schema_versions);

  // ddl related
  int create_resource_unit(const obrpc::ObCreateResourceUnitArg& arg);
  int alter_resource_unit(const obrpc::ObAlterResourceUnitArg& arg);
  int drop_resource_unit(const obrpc::ObDropResourceUnitArg& arg);
  int create_resource_pool(const obrpc::ObCreateResourcePoolArg& arg);
  int alter_resource_pool(const obrpc::ObAlterResourcePoolArg& arg);
  int drop_resource_pool(const obrpc::ObDropResourcePoolArg& arg);
  int split_resource_pool(const obrpc::ObSplitResourcePoolArg& arg);
  int merge_resource_pool(const obrpc::ObMergeResourcePoolArg& arg);
  int create_tenant(const obrpc::ObCreateTenantArg& arg, obrpc::UInt64& tenant_id);
  int create_tenant_end(const obrpc::ObCreateTenantEndArg& arg);
  int commit_alter_tenant_locality(const rootserver::ObCommitAlterTenantLocalityArg& arg);
  int commit_alter_tablegroup_locality(const rootserver::ObCommitAlterTablegroupLocalityArg& arg);
  int commit_alter_table_locality(const rootserver::ObCommitAlterTableLocalityArg& arg);
  int drop_tenant(const obrpc::ObDropTenantArg& arg);
  int flashback_tenant(const obrpc::ObFlashBackTenantArg& arg);
  int purge_tenant(const obrpc::ObPurgeTenantArg& arg);
  int modify_tenant(const obrpc::ObModifyTenantArg& arg);
  int lock_tenant(const obrpc::ObLockTenantArg& arg);
  int modify_system_variable(const obrpc::ObModifySysVarArg& arg);
  int add_system_variable(const obrpc::ObAddSysVarArg& arg);
  int create_database(const obrpc::ObCreateDatabaseArg& arg, obrpc::UInt64& db_id);
  int create_tablegroup(const obrpc::ObCreateTablegroupArg& arg, obrpc::UInt64& tg_id);
  int get_frozen_status(const obrpc::Int64& arg, storage::ObFrozenStatus& frozen_status);
  int create_table(const obrpc::ObCreateTableArg& arg, obrpc::UInt64& table_id);
  int alter_database(const obrpc::ObAlterDatabaseArg& arg);
  int alter_table(const obrpc::ObAlterTableArg& arg, obrpc::ObAlterTableRes& res);
  int alter_tablegroup(const obrpc::ObAlterTablegroupArg& arg);
  int rename_table(const obrpc::ObRenameTableArg& arg);
  int truncate_table(const obrpc::ObTruncateTableArg& arg);
  int create_index(const obrpc::ObCreateIndexArg& arg, obrpc::ObAlterTableRes& res);
  int drop_table(const obrpc::ObDropTableArg& arg);
  int drop_database(const obrpc::ObDropDatabaseArg& arg, obrpc::UInt64& affected_row);
  int drop_tablegroup(const obrpc::ObDropTablegroupArg& arg);
  int drop_index(const obrpc::ObDropIndexArg& arg);
  int rebuild_index(const obrpc::ObRebuildIndexArg& arg, obrpc::ObAlterTableRes& res);
  // the interface only for switchover: execute skip check enable_ddl
  int force_drop_index(const obrpc::ObDropIndexArg& arg);
  int flashback_index(const obrpc::ObFlashBackIndexArg& arg);
  int purge_index(const obrpc::ObPurgeIndexArg& arg);
  int create_table_like(const obrpc::ObCreateTableLikeArg& arg);
  int refresh_config();
  int get_frozen_version(obrpc::Int64& frozen_version);
  int root_major_freeze(const obrpc::ObRootMajorFreezeArg& arg);
  int root_minor_freeze(const obrpc::ObRootMinorFreezeArg& arg);
  int update_index_status(const obrpc::ObUpdateIndexStatusArg& arg);
  int purge_table(const obrpc::ObPurgeTableArg& arg);
  int flashback_table_from_recyclebin(const obrpc::ObFlashBackTableFromRecyclebinArg& arg);
  int flashback_table_to_time_point(const obrpc::ObFlashBackTableToScnArg& arg);
  int purge_database(const obrpc::ObPurgeDatabaseArg& arg);
  int flashback_database(const obrpc::ObFlashBackDatabaseArg& arg);
  int check_tenant_in_alter_locality(const uint64_t tenant_id, bool& in_alter_locality);
  int root_split_partition(const obrpc::ObRootSplitPartitionArg& arg);

  int create_restore_point(const obrpc::ObCreateRestorePointArg& arg);
  int drop_restore_point(const obrpc::ObDropRestorePointArg& arg);
  // for inner table monitor, purge in fixed time
  int purge_expire_recycle_objects(const obrpc::ObPurgeRecycleBinArg& arg, obrpc::Int64& affected_rows);
  int check_unique_index_response(const obrpc::ObCheckUniqueIndexResponseArg& arg);
  int calc_column_checksum_repsonse(const obrpc::ObCalcColumnChecksumResponseArg& arg);
  int optimize_table(const obrpc::ObOptimizeTableArg& arg);

  //----Functions for managing privileges----
  int create_user(obrpc::ObCreateUserArg& arg, common::ObSArray<int64_t>& failed_index);
  int drop_user(const obrpc::ObDropUserArg& arg, common::ObSArray<int64_t>& failed_index);
  int rename_user(const obrpc::ObRenameUserArg& arg, common::ObSArray<int64_t>& failed_index);
  int set_passwd(const obrpc::ObSetPasswdArg& arg);
  int grant(const obrpc::ObGrantArg& arg);
  int revoke_user(const obrpc::ObRevokeUserArg& arg);
  int lock_user(const obrpc::ObLockUserArg& arg, common::ObSArray<int64_t>& failed_index);
  int revoke_database(const obrpc::ObRevokeDBArg& arg);
  int revoke_table(const obrpc::ObRevokeTableArg& arg);
  int revoke_syspriv(const obrpc::ObRevokeSysPrivArg& arg);
  int alter_user_profile(const obrpc::ObAlterUserProfileArg& arg);
  //----End of functions for managing privileges----

  //----Functions for managing outlines----
  int create_outline(const obrpc::ObCreateOutlineArg& arg);
  int alter_outline(const obrpc::ObAlterOutlineArg& arg);
  int drop_outline(const obrpc::ObDropOutlineArg& arg);
  //----End of functions for managing outlines----

  //----Functions for managing UDF----
  int create_user_defined_function(const obrpc::ObCreateUserDefinedFunctionArg& arg);
  int drop_user_defined_function(const obrpc::ObDropUserDefinedFunctionArg& arg);
  //----End of functions for managing UDF----
  //----Functions for managing dblinks----
  int create_dblink(const obrpc::ObCreateDbLinkArg& arg);
  int drop_dblink(const obrpc::ObDropDbLinkArg& arg);
  //----End of functions for managing dblinks----

  //----Functions for managing synonyms----
  int create_synonym(const obrpc::ObCreateSynonymArg& arg);
  int drop_synonym(const obrpc::ObDropSynonymArg& arg);
  //----End of functions for managing synonyms----

  //----Functions for managing sequence----
  // create alter drop actions all in one, avoid noodle-like code
  int do_sequence_ddl(const obrpc::ObSequenceDDLArg& arg);
  //----End of functions for managing sequence----

  //----Functions for managing profile----
  int do_profile_ddl(const obrpc::ObProfileDDLArg& arg);
  //----End of functions for managing sequence----

  // server related
  int add_server(const obrpc::ObAdminServerArg& arg);
  int delete_server(const obrpc::ObAdminServerArg& arg);
  int cancel_delete_server(const obrpc::ObAdminServerArg& arg);
  int start_server(const obrpc::ObAdminServerArg& arg);
  int stop_server(const obrpc::ObAdminServerArg& arg);

  // zone related
  int add_zone(const obrpc::ObAdminZoneArg& arg);
  int delete_zone(const obrpc::ObAdminZoneArg& arg);
  int start_zone(const obrpc::ObAdminZoneArg& arg);
  int stop_zone(const obrpc::ObAdminZoneArg& arg);
  int alter_zone(const obrpc::ObAdminZoneArg& arg);

  // gts replica stop server/zone related
  int check_gts_replica_enough_when_stop_server(
      const obrpc::ObCheckGtsReplicaStopServer& server_need_stopped, obrpc::Bool& can_stop);
  int check_gts_replica_enough_when_stop_zone(
      const obrpc::ObCheckGtsReplicaStopZone& zone_need_stopped, obrpc::Bool& can_stop);

  // system admin command (alter system ...)
  int admin_switch_replica_role(const obrpc::ObAdminSwitchReplicaRoleArg& arg);
  int admin_switch_rs_role(const obrpc::ObAdminSwitchRSRoleArg& arg);
  int admin_drop_replica(const obrpc::ObAdminDropReplicaArg& arg);
  int admin_change_replica(const obrpc::ObAdminChangeReplicaArg& arg);
  int admin_migrate_replica(const obrpc::ObAdminMigrateReplicaArg& arg);
  int admin_report_replica(const obrpc::ObAdminReportReplicaArg& arg);
  int admin_recycle_replica(const obrpc::ObAdminRecycleReplicaArg& arg);
  int admin_merge(const obrpc::ObAdminMergeArg& arg);
  int admin_clear_roottable(const obrpc::ObAdminClearRoottableArg& arg);
  int admin_refresh_schema(const obrpc::ObAdminRefreshSchemaArg& arg);
  int admin_set_config(obrpc::ObAdminSetConfigArg& arg);
  int admin_clear_location_cache(const obrpc::ObAdminClearLocationCacheArg& arg);
  int admin_reload_gts();
  int admin_refresh_memory_stat(const obrpc::ObAdminRefreshMemStatArg& arg);
  int admin_reload_unit();
  int admin_reload_server();
  int admin_reload_zone();
  int admin_clear_merge_error();
  int admin_migrate_unit(const obrpc::ObAdminMigrateUnitArg& arg);
  int admin_upgrade_virtual_schema();
  int run_job(const obrpc::ObRunJobArg& arg);
  int run_upgrade_job(const obrpc::ObUpgradeJobArg& arg);
  int admin_flush_cache(const obrpc::ObAdminFlushCacheArg& arg);
  int admin_upgrade_cmd(const obrpc::Bool& arg);
  int admin_rolling_upgrade_cmd(const obrpc::ObAdminRollingUpgradeArg& arg);
  int admin_set_tracepoint(const obrpc::ObAdminSetTPArg& arg);
  int restore_tenant(const obrpc::ObRestoreTenantArg& arg);
  /* physical restore */
  int physical_restore_tenant(const obrpc::ObPhysicalRestoreTenantArg& arg);
  int rebuild_index_in_restore(const obrpc::ObRebuildIndexInRestoreArg& arg);
  /*-----------------*/
  int admin_load_baseline(const obrpc::ObAdminLoadBaselineArg& arg);
  int refresh_time_zone_info(const obrpc::ObRefreshTimezoneArg& arg);
  int request_time_zone_info(const common::ObRequestTZInfoArg& arg, common::ObRequestTZInfoResult& result);
  // async tasks and callbacks
  // @see ObStatusChangeCallback
  int submit_update_all_server_task(const common::ObAddr& server);
  int submit_start_server_task(const common::ObAddr& server);
  int submit_stop_server_task(const common::ObAddr& server);
  int submit_offline_server_task(const common::ObAddr& server);
  int submit_report_core_table_replica_task();
  int submit_reload_unit_manager_task();
  int report_replica();
  int report_single_replica(const common::ObPartitionKey& key);
  // @see RsListChangeCb
  int submit_update_rslist_task(const bool force_update = false);
  // @see ObIMergeErrorCb
  int submit_merge_error_task();
  int submit_schema_split_task();
  int submit_schema_split_task_v2();
  int submit_upgrade_task(const int64_t version);
  int submit_upgrade_storage_format_version_task();
  int submit_create_inner_schema_task();
  int submit_schema_revise_task();
  int submit_statistic_primary_zone_count();
  int submit_create_ha_gts_util();
  int submit_async_minor_freeze_task(const obrpc::ObRootMinorFreezeArg& arg);
  // @see ObLostReplicaCheckerTask
  int submit_lost_replica_checker_task();
  int submit_update_all_server_config_task();

  int submit_index_sstable_build_task(const sql::ObIndexSSTableBuilder::BuildIndexJob& job,
      sql::ObIndexSSTableBuilder::ReplicaPicker& replica_picker, const int64_t abs_timeout_us);
  // may modify arg before taking effect
  int set_config_pre_hook(obrpc::ObAdminSetConfigArg& arg);
  // arg is readonly after take effect
  int set_config_post_hook(const obrpc::ObAdminSetConfigArg& arg);

  // @see ObReplicaControlCleanTask
  int submit_replica_control_clean_task();

  // @see ObDelSvrWorkingDirCleanTask
  int submit_del_svr_working_dir_clean_task();

  // @see ObBlkRecycleScheduleTask
  int submit_blk_recycle_schedule_task();

  //@see ObBlacklistProcess
  int submit_black_list_task(const int64_t count, const common::ObIArray<uint64_t>& table_ids,
      const common::ObIArray<int64_t>& partition_ids, const common::ObIArray<common::ObAddr>& servers,
      const share::ObPartitionReplica::FailList& fail_msgs);

  // @see ObRestartTask
  int after_restart();
  int do_after_full_service();
  int schedule_restart_timer_task(const int64_t delay);
  int schedule_self_check_task();
  int schedule_temporary_offline_timer_task();
  // @see ObCheckServerTask
  int schedule_check_server_timer_task();
  // @see ObRefreshServerTask
  int schedule_refresh_server_timer_task(const int64_t delay);
  // @see ObInnerTableMonitorTask
  int schedule_inner_table_monitor_task();
  int schedule_recyclebin_task(int64_t delay);
  int schedule_force_drop_schema_task(int64_t delay);
  // @see ObInspector
  int schedule_inspector_task();
  int schedule_update_rs_list_task();
  // update statistic cache
  int update_stat_cache(const obrpc::ObUpdateStatCacheArg& arg);

  int schedule_load_building_index_task();
  // ob_admin command, must be called in ddl thread
  int force_create_sys_table(const obrpc::ObForceCreateSysTableArg& arg);
  int force_set_locality(const obrpc::ObForceSetLocalityArg& arg);

  int restore_partitions(const obrpc::ObRestorePartitionsArg& arg);
  int physical_restore_partitions(const obrpc::ObRestorePartitionsArg& arg);
  int logical_restore_partitions(const obrpc::ObRestorePartitionsArg& arg);

  int check_is_log_sync(bool& is_log_sync, const common::ObIArray<common::ObAddr>& stop_server);
  int generate_stop_server_log_in_sync_dest_server_array(const common::ObIArray<common::ObAddr>& alive_server_array,
      const common::ObIArray<common::ObAddr>& excluded_server_array,
      common::ObIArray<common::ObAddr>& dest_server_array);
  int generate_log_in_sync_dest_server_buff(common::ObIAllocator& allocator,
      const common::ObIArray<common::ObAddr>& dest_server_array, common::ObString& dest_server_buff);
  int log_nop_operation(const obrpc::ObDDLNopOpreatorArg& arg);
  int broadcast_schema(const obrpc::ObBroadcastSchemaArg& arg);
  int check_other_rs_exist(bool& is_other_rs_exist);
  ObRsGtsManager& get_rs_gts_manager()
  {
    return rs_gts_manager_;
  }
  ObFreezeInfoManager& get_freeze_manager()
  {
    return freeze_info_manager_;
  }
  ObDDLService& get_ddl_service()
  {
    return ddl_service_;
  }

  int get_schema_split_version(int64_t& version_in_core_table, int64_t& version_in_ddl_table);
  int finish_schema_split(const obrpc::ObFinishSchemaSplitArg& arg);

  int check_merge_finish(const obrpc::ObCheckMergeFinishArg& arg);
  int get_recycle_schema_versions(
      const obrpc::ObGetRecycleSchemaVersionsArg& arg, obrpc::ObGetRecycleSchemaVersionsResult& result);
  // backup and restore
  int force_drop_schema(const obrpc::ObForceDropSchemaArg& arg);
  int handle_archive_log(const obrpc::ObArchiveLogArg& arg);
  int handle_backup_database(const obrpc::ObBackupDatabaseArg& arg);
  int handle_backup_manage(const obrpc::ObBackupManageArg& arg);
  int handle_validate_database(const obrpc::ObBackupManageArg& arg);
  int handle_validate_backupset(const obrpc::ObBackupManageArg& arg);
  int handle_cancel_validate(const obrpc::ObBackupManageArg& arg);
  int update_table_schema_version(const obrpc::ObUpdateTableSchemaVersionArg& arg);
  int modify_schema_in_restore(const obrpc::ObRestoreModifySchemaArg& arg);
  int check_backup_scheduler_working(obrpc::Bool& is_working);
  int update_freeze_schema_versions(
      const int64_t frozen_version, obrpc::ObTenantSchemaVersions& tenant_schema_versions);
  int send_physical_restore_result(const obrpc::ObPhysicalRestoreResult& res);
  int execute_range_part_split(const obrpc::ObExecuteRangePartSplitArg& arg);
  int get_is_in_bootstrap(bool& is_bootstrap) const;
  int purge_recyclebin_objects(int64_t purge_each_time);
  /////////////////////////////////////////////////////
  // for standby
  int switch_cluster(const obrpc::ObAlterClusterInfoArg& arg);
  int alter_cluster_attr(const obrpc::ObAlterClusterInfoArg& arg);
  int alter_cluster_info(const obrpc::ObAlterClusterInfoArg& arg);
  int cluster_regist(const obrpc::ObRegistClusterArg& arg, obrpc::ObRegistClusterRes& res);
  int alter_standby(const obrpc::ObAdminClusterArg& arg);
  int get_cluster_info(const obrpc::ObGetClusterInfoArg& arg, share::ObClusterInfo& cluster_info);
  int get_switchover_status(obrpc::ObGetSwitchoverStatusRes& res);
  int check_cluster_valid_to_add(const obrpc::ObCheckAddStandbyArg& arg);
  int check_standby_can_access(const obrpc::ObCheckStandbyCanAccessArg& arg, obrpc::Bool& can_access);
  int get_schema_snapshot(const obrpc::ObSchemaSnapshotArg& arg, obrpc::ObSchemaSnapshotRes& res);
  int cluster_action_verify(const obrpc::ObClusterActionVerifyArg& arg);
  int gen_next_schema_version(const obrpc::ObDDLArg& arg);
  int get_cluster_stats(obrpc::ObClusterTenantStats& tenant_stats);
  int standby_upgrade_virtual_schema(const obrpc::ObDDLNopOpreatorArg& arg);
  int update_standby_cluster_info(const share::ObClusterAddr& arg);
  int cluster_heartbeat(const share::ObClusterAddr& arg, obrpc::ObStandbyHeartBeatRes& res);
  int standby_grant(const obrpc::ObStandbyGrantArg& arg);
  int finish_replay_schema(const obrpc::ObFinishReplayArg& arg);

  ////////////////////////////////////////////////////
G
gm 已提交
1227
private:
O
oceanbase-admin 已提交
1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287
  int check_parallel_ddl_conflict(share::schema::ObSchemaGetterGuard& schema_guard, const obrpc::ObDDLArg& arg);
  int check_can_start_as_primary();
  int fetch_root_partition_info();
  int generate_user(const ObClusterType& cluster_type, const char* usr_name, const char* user_passwd);
  int set_core_table_unit_id();
  // create system table in mysql backend for debugging mode.
  int init_debug_database();
  int do_restart();
  int refresh_server(const bool fast_recover, const bool need_retry);
  int refresh_schema(const bool fast_recover);
  int init_sequence_id();
  int load_server_manager();
  int start_timer_tasks();
  int stop_timer_tasks();
  int request_heartbeats();
  int self_check();
  int update_all_server_and_rslist();
  int check_zone_and_server(const ObIArray<ObAddr>& servers, bool& is_same_zone, bool& is_all_stopped);
  int update_rslist();
  int check_can_stop(
      const common::ObZone& zone, const common::ObIArray<common::ObAddr>& servers, const bool is_stop_zone);
  bool have_other_stop_task(const ObZone& zone);
  int check_is_log_sync_twice(bool& is_log_sync, const common::ObIArray<common::ObAddr>& stop_server);
  int init_sys_admin_ctx(ObSystemAdminCtx& ctx);
  int set_cluster_version();
  int get_root_partition_table(common::ObArray<common::ObAddr>& rs_list);
  bool is_replica_count_reach_rs_limit(int64_t replica_count)
  {
    return replica_count > OB_MAX_CLUSTER_REPLICA_COUNT;
  }
  int update_all_server_config();
  int get_readwrite_servers(
      const common::ObIArray<common::ObAddr>& input_servers, common::ObIArray<common::ObAddr>& readwrite_servers);
  int check_tenant_group_config_legality(
      common::ObIArray<ObTenantGroupParser::TenantNameGroup>& tenant_groups, bool& legal);
  int generate_table_schema_in_tenant_space(
      const obrpc::ObCreateTableArg& arg, share::schema::ObTableSchema& table_schema);
  int clear_bootstrap();
  int check_tenant_gts_config(
      const int64_t tenant_id, bool& tenant_gts_config_ok, share::schema::ObSchemaGetterGuard& schema_guard);
  int check_database_config(
      const int64_t tenant_id, bool& db_config_ok, share::schema::ObSchemaGetterGuard& schema_guard);
  int check_table_config(const int64_t tenant_id, bool& table_config_ok, bool& table_split_ok,
      share::schema::ObSchemaGetterGuard& schema_guard, const int64_t snapshot_schema_version);
  int check_tablegroup_config(const int64_t tenant_id, bool& tablegroup_config_ok, bool& tablegroup_split_ok,
      share::schema::ObSchemaGetterGuard& schema_guard, const int64_t snapshot_schema_version);
  int check_all_tenant_config(ObArray<ObString>& not_allow_reasons, share::schema::ObSchemaGetterGuard& schema_guard,
      const int64_t snapshot_schema_version, ObArray<uint64_t>& tenant_ids);
  bool continue_check(const int ret);
  int handle_backup_database_cancel(const obrpc::ObBackupManageArg& arg);
  int handle_backup_delete_obsolete_backup(const obrpc::ObBackupManageArg& arg);
  int handle_backup_delete_backup_set(const obrpc::ObBackupManageArg& arg);
  int refresh_unit_replica_counter();
  int reload_unit_replica_counter_task();
  inline static bool cmp_tenant_id(const uint64_t lhs, const uint64_t tenant_id)
  {
    return lhs < tenant_id;
  }
  int handle_cancel_delete_backup(const obrpc::ObBackupManageArg& arg);

G
gm 已提交
1288
private:
O
oceanbase-admin 已提交
1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300
  bool is_sys_tenant(const common::ObString& tenant_name);
  int table_is_split(
      uint64_t tenant_id, const ObString& database_name, const ObString& table_name, const bool is_index = false);
  int table_allow_ddl_operation(const obrpc::ObAlterTableArg& arg);
  int get_table_schema(uint64_t tenant_id, const common::ObString& database_name, const common::ObString& table_name,
      const bool is_index, const int64_t session_id, const share::schema::ObTableSchema*& table_schema);
  int batch_root_split_partition(const obrpc::ObRootSplitPartitionArg& arg);
  int update_baseline_schema_version();
  int finish_bootstrap();
  void construct_lease_expire_time(const share::ObLeaseRequest& lease_request, share::ObLeaseResponse& lease_response,
      const share::ObServerStatus& server_status);

G
gm 已提交
1301
private:
O
oceanbase-admin 已提交
1302 1303 1304 1305 1306 1307 1308
  int update_all_sys_tenant_schema_version();
  int finish_schema_split_v1(const obrpc::ObFinishSchemaSplitArg& arg);
  int finish_schema_split_v2(const obrpc::ObFinishSchemaSplitArg& arg);
  int check_has_restore_tenant(bool& has_restore_tenant);
  int build_range_part_split_arg(const common::ObPartitionKey& partition_key, const common::ObRowkey& rowkey,
      share::schema::AlterTableSchema& alter_table_schema);

G
gm 已提交
1309
private:
O
oceanbase-admin 已提交
1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459
  static const int64_t OB_MAX_CLUSTER_REPLICA_COUNT = 10000000;
  static const int64_t OB_ROOT_SERVICE_START_FAIL_COUNT_UPPER_LIMIT = 10;
  bool inited_;
  volatile bool server_refreshed_;  // server manager reload and force request heartbeat
  // use mysql server backend for debug.
  bool debug_;

  common::ObAddr self_addr_;
  common::ObServerConfig* config_;
  common::ObConfigManager* config_mgr_;

  obrpc::ObSrvRpcProxy rpc_proxy_;
  obrpc::ObCommonRpcProxy common_proxy_;
  common::ObMySQLProxy sql_proxy_;
  common::ObOracleSqlProxy oracle_sql_proxy_;
  share::ObRemoteSqlProxy* remote_sql_proxy_;
  observer::ObRestoreCtx* restore_ctx_;
  share::ObRsMgr* rs_mgr_;
  share::schema::ObMultiVersionSchemaService* schema_service_;

  // server manager related
  ObStatusChangeCallback status_change_cb_;
  ObServerChangeCallback server_change_callback_;
  ObServerManager server_manager_;
  ObHeartbeatChecker hb_checker_;
  ObAllServerChecker server_checker_;
  RsListChangeCb rs_list_change_cb_;

  // major freeze
  ObFreezeInfoManager freeze_info_manager_;
  ObRootMajorFreezeV2 root_major_freeze_v2_;

  // minor freeze
  ObRootMinorFreeze root_minor_freeze_;

  // partition table related
  share::ObPartitionTableOperator* pt_operator_;
  share::ObRemotePartitionTableOperator* remote_pt_operator_;
  share::ObIPartitionLocationCache* location_cache_;

  ObZoneManager zone_manager_;

  // ddl related
  ObDDLService ddl_service_;
  ObUnitManager unit_manager_;
  ObLeaderCoordinator leader_coordinator_;

  ObPartitionTableUtil pt_util_;
  ObDailyMergeScheduler daily_merge_scheduler_;
  MergeErrorCallback merge_error_cb_;

  ObMajorFreezeLauncher major_freeze_launcher_;
  ObFreezeInfoUpdater freeze_info_updater_;

  ObGlobalMaxDecidedTransVersionMgr global_trans_version_mgr_;

  // partition balance
  ObRebalanceTaskExecutor rebalance_task_executor_;
  ObRebalanceTaskMgr rebalance_task_mgr_;
  ObRootBalancer root_balancer_;

  // empty server checker
  ObEmptyServerChecker empty_server_checker_;

  // lost replica checker
  ObLostReplicaChecker lost_replica_checker_;

  // thread checker
  ObRsThreadChecker thread_checker_;

  // virtual table related
  ObVTableLocationGetter vtable_location_getter_;

  // broadcast rs list related
  share::ObUnifiedAddrAgent addr_agent_;

  // upgrade inspection
  ObRootInspection root_inspection_;

  // schema split
  ObSchemaSplitExecutor schema_split_executor_;

  // upgrade post job
  ObUpgradeExecutor upgrade_executor_;

  // upgrade storage format version
  ObUpgradeStorageFormatVersionExecutor upgrade_storage_format_executor_;

  // create inner oracle role(for upgrade)
  ObCreateInnerSchemaExecutor create_inner_schema_executor_;

  // revise data in inner_table (for upgrade)
  ObSchemaReviseExecutor schema_revise_executor_;

  // avoid concurrent run of do_restart and bootstrap
  common::ObLatch bootstrap_lock_;

  common::SpinRWLock broadcast_rs_list_lock_;

  // Inner table mointor
  ObInnerTableMonitor inner_table_monitor_;
  ObPartitionChecksumChecker partition_checksum_checker_;

  // the single task queue for all async tasks and timer tasks
  common::ObWorkQueue task_queue_;
  common::ObWorkQueue inspect_task_queue_;

  // async timer tasks
  ObRestartTask restart_task_;                           // not repeat & no retry
  ObRefreshServerTask refresh_server_task_;              // not repeat & no retry
  ObCheckServerTask check_server_task_;                  // repeat & no retry
  ObSelfCheckTask self_check_task_;                      // repeat to succeed & no retry
  ObLoadIndexBuildTask load_building_index_task_;        // repeat to succeed & no retry
  share::ObEventTableClearTask event_table_clear_task_;  // repeat & no retry
  ObInnerTableMonitorTask inner_table_monitor_task_;     // repeat & no retry

  ObInspector inspector_task_;                    // repeat & no retry
  ObPurgeRecyclebinTask purge_recyclebin_task_;   // not repeat & no retry
  ObForceDropSchemaTask force_drop_schema_task_;  // not repeat & no retry
  // for set_config
  ObLatch set_config_lock_;

  ObGlobalIndexBuilder global_index_builder_;
  ObPartitionSpliter partition_spliter_;
  ObSnapshotInfoManager snapshot_manager_;
  int64_t core_meta_table_version_;
  ObRsGtsManager rs_gts_manager_;
  ObUpdateRsListTimerTask update_rs_list_timer_task_;
  ObRsGtsTaskMgr rs_gts_task_mgr_;
  ObRsGtsMonitor rs_gts_monitor_;
  int64_t baseline_schema_version_;
  mutable common::SpinRWLock upgrade_cluster_create_ha_gts_lock_;

  // backup
  ObRestoreScheduler restore_scheduler_;
  ObLogArchiveScheduler log_archive_scheduler_;
  ObRootBackup root_backup_;
  ObRootValidate root_validate_;
  int64_t start_service_time_;
  ObRsStatus rs_status_;

  int64_t fail_count_;
  ObSchemaHistoryRecycler schema_history_recycler_;
  ObBackupDataClean backup_data_clean_;
  ObBackupAutoDeleteExpiredData backup_auto_delete_;
  ObReloadUnitReplicaCounterTask reload_unit_replica_counter_task_;
  ObSinglePartBalance single_part_balance_;
  ObBackupLeaseService backup_lease_service_;
  ObRestorePointService restore_point_service_;

G
gm 已提交
1460
private:
O
oceanbase-admin 已提交
1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537
  DISALLOW_COPY_AND_ASSIGN(ObRootService);
};

template <typename TASK, typename T>
int ObRootService::do_receive_balance_over(TASK& task, const T& arg)
{
  int ret = common::OB_SUCCESS;
  RS_LOG(INFO, "receive balance over", K(arg));
  common::ObArray<int> rc_array;
  if (!inited_) {
    ret = common::OB_NOT_INIT;
    RS_LOG(WARN, "not init", K(ret));
  } else if (!arg.is_valid()) {
    ret = common::OB_INVALID_ARGUMENT;
    RS_LOG(WARN, "invalid arg", K(arg), K(ret));
  } else if (OB_FAIL(task.build_by_task_result(arg))) {
    RS_LOG(WARN, "fail to init task", K(ret), K(arg));
  } else if (OB_FAIL(rc_array.push_back(static_cast<int>(arg.result_)))) {
    RS_LOG(WARN, "fail to push back", K(ret));
  } else if (OB_FAIL(try_block_server(static_cast<int>(arg.result_), arg.dst_.get_server()))) {
    RS_LOG(WARN, "try block failed", "rc", arg.result_, "server", arg.dst_.get_server(), K(ret));
  } else if (OB_FAIL(rebalance_task_mgr_.execute_over(task, rc_array))) {
    RS_LOG(WARN, "execute over failed", K(ret), K(arg), K(task));
  } else {
    ObRebalanceTaskType type = task.get_rebalance_task_type();
    if (ObRebalanceTaskType::COPY_SSTABLE == type || ObRebalanceTaskType::PHYSICAL_RESTORE_REPLICA == type) {
      restore_scheduler_.wakeup();
    }
    RS_LOG(INFO, "process balance rpc over", K(arg), K(task));
  }
  return ret;
}

template <typename TASK, typename T>
int ObRootService::do_receive_batch_balance_over(TASK& task, const T& arg)
{
  int ret = common::OB_SUCCESS;
  RS_LOG(INFO, "receive batch balance over", K(arg));
  if (!inited_) {
    ret = common::OB_NOT_INIT;
    RS_LOG(WARN, "not init", K(ret));
  } else if (!arg.is_valid()) {
    ret = common::OB_INVALID_ARGUMENT;
    RS_LOG(WARN, "invalid arg", K(ret), K(arg));
  } else if (OB_FAIL(task.build_by_task_result(arg))) {
    RS_LOG(WARN, "fail to init task", K(ret), K(arg));
  } else {
    common::ObArray<int> rc_array;
    for (int64_t i = 0; OB_SUCC(ret) && i < arg.res_array_.count(); ++i) {
      if (OB_FAIL(try_block_server(
              static_cast<int>(arg.res_array_.at(i).result_), arg.res_array_.at(i).dst_.get_server()))) {
        RS_LOG(WARN,
            "try block server failed",
            K(ret),
            "server",
            arg.res_array_.at(i).dst_.get_server(),
            "ret_code",
            arg.res_array_.at(i).result_);
      } else if (OB_FAIL(rc_array.push_back(static_cast<int>(arg.res_array_.at(i).result_)))) {
        RS_LOG(WARN, "fail to push back", K(ret));
      } else {
      }  // no more to do
    }
    if (OB_SUCC(ret)) {
      if (OB_FAIL(rebalance_task_mgr_.execute_over(task, rc_array))) {
        RS_LOG(WARN, "fail to execute over", K(ret), K(task));
      } else {
      }  // no more to do
    }
  }
  return ret;
}

}  // end namespace rootserver
}  // end namespace oceanbase

#endif  // OCEANBASE_ROOTSERVER_OB_ROOT_SERVICE_H_