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 SERVER
14

15
#include "observer/mysql/obmp_statistic.h"
16
#include "observer/mysql/obmp_utils.h"
17
#include "rpc/obmysql/ob_mysql_packet.h"
18
#include "rpc/obmysql/packet/ompk_string.h"
19
#include "rpc/obmysql/obsm_struct.h"
20

21
namespace oceanbase
22
{
23
using namespace common;
24
using namespace obmysql;
25

26
namespace observer
27
{
28
int ObMPStatistic::process()
29
{
30
  int ret = common::OB_SUCCESS;
31
  bool need_disconnect = true;
32
  bool need_response_error = true;
33
  //Attention::it is BUG when using like followers (build with release):
34
  //  obmysql::OMPKString pkt(ObString("Active threads not support"));
35
  //
36
  const common::ObString tmp_string("Active threads not support");
37
  obmysql::OMPKString pkt(tmp_string);
38
  ObSMConnection *conn = NULL;
39
  const ObMySQLRawPacket &mysql_pkt = reinterpret_cast<const ObMySQLRawPacket&>(req_->get_packet());
40

41
  if (OB_FAIL(packet_sender_.alloc_ezbuf())) {
42
    LOG_WARN("failed to alloc easy buf", K(ret));
43
  } else if (OB_FAIL(packet_sender_.update_last_pkt_pos())) {
44
    LOG_WARN("failed to update last packet pos", K(ret));
45
  } else if (OB_ISNULL(conn = get_conn())) {
46
    ret = OB_ERR_UNEXPECTED;
47
    LOG_WARN("get connection fail", K(conn), K(ret));
48
  } else if (conn->proxy_cap_flags_.is_extra_ok_packet_for_statistics_support()) {
49
    sql::ObSQLSessionInfo *session = NULL;
50
    if (OB_FAIL(get_session(session))) {
51
      LOG_WARN("get session fail", K(ret));
52
    } else if (OB_ISNULL(session)) {
53
      ret = OB_ERR_UNEXPECTED;
54
      LOG_WARN("sql session info is null", K(ret));
55
    } else if (OB_FAIL(process_kill_client_session(*session))) {
56
      LOG_WARN("client session has been killed", K(ret));
57
    } else if (FALSE_IT(session->set_txn_free_route(mysql_pkt.txn_free_route()))) {
58
    } else if (OB_FAIL(process_extra_info(*session, mysql_pkt, need_response_error))) {
59
      LOG_WARN("fail get process extra info", K(ret));
60
    } else if (FALSE_IT(session->post_sync_session_info())) {
61
    } else if (OB_FAIL(update_transmission_checksum_flag(*session))) {
62
      LOG_WARN("update transmisson checksum flag failed", K(ret));
63
    } else {
64
      ObOKPParam ok_param; // use default values
65
      if (OB_FAIL(send_ok_packet(*session, ok_param, &pkt))) {
66
        LOG_WARN("fail to send ok pakcet in statistic response", K(ok_param), K(ret));
67
      }
68
    }
69
    if (OB_LIKELY(NULL != session)) {
70
      revert_session(session);
71
    }
72
  } else if (OB_FAIL(response_packet(pkt, NULL))) {
73
    RPC_OBMYSQL_LOG(WARN, "fail to response statistic packet", K(ret));
74
  } else {
75
    // do nothing
76
  }
77

78
  if (OB_FAIL(ret) && need_response_error) {
79
    send_error_packet(ret, NULL);
80
  }
81
  if (OB_FAIL(ret) && need_disconnect) {
82
    force_disconnect();
83
    LOG_WARN("disconnect connection", KR(ret));
84
  }
85
  return ret;
86
}
87

88

89
} // namespace observer
90
} // namespace oceanbase
91