| 1092 | } |
| 1093 | |
| 1094 | Status ClientRequestState::ExecShutdownRequest() { |
| 1095 | const TShutdownParams& request = exec_request().admin_request.shutdown_params; |
| 1096 | bool backend_port_specified = request.__isset.backend && request.backend.port != 0; |
| 1097 | int port = backend_port_specified ? request.backend.port : FLAGS_krpc_port; |
| 1098 | // Use the local shutdown code path if the host is unspecified or if it exactly matches |
| 1099 | // the configured host/port. This avoids the possibility of RPC errors preventing |
| 1100 | // shutdown. |
| 1101 | if (!request.__isset.backend |
| 1102 | || (request.backend.hostname == FLAGS_hostname && port == FLAGS_krpc_port)) { |
| 1103 | ShutdownStatusPB shutdown_status; |
| 1104 | int64_t deadline_s = request.__isset.deadline_s ? request.deadline_s : -1; |
| 1105 | RETURN_IF_ERROR(parent_server_->StartShutdown(deadline_s, &shutdown_status)); |
| 1106 | SetResultSet({ImpalaServer::ShutdownStatusToString(shutdown_status)}); |
| 1107 | return Status::OK(); |
| 1108 | } |
| 1109 | |
| 1110 | // KRPC relies on resolved IP address, so convert hostname. |
| 1111 | IpAddr ip_address; |
| 1112 | Status ip_status = HostnameToIpAddr(request.backend.hostname, &ip_address); |
| 1113 | if (!ip_status.ok()) { |
| 1114 | VLOG(1) << "Could not convert hostname " << request.backend.hostname |
| 1115 | << " to ip address, error: " << ip_status.GetDetail(); |
| 1116 | return ip_status; |
| 1117 | } |
| 1118 | // Find BackendId for the given remote ip address and port from cluster membership. |
| 1119 | // The searching is not efficient, but Shutdown Requests are not called frequently. |
| 1120 | // The BackendId is used to generate UDS address for Unix domain socket. Leave the |
| 1121 | // Id value as 0 if it's not found in cluster membership. |
| 1122 | // Note that UDS is only used when FLAGS_rpc_use_unix_domain_socket is set as true. |
| 1123 | UniqueIdPB backend_id; |
| 1124 | backend_id.set_hi(0); |
| 1125 | backend_id.set_lo(0); |
| 1126 | if (ExecEnv::GetInstance()->rpc_mgr()->IsKrpcUsingUDS()) { |
| 1127 | if (ExecEnv::GetInstance()->rpc_mgr()->GetUdsAddressUniqueId() |
| 1128 | == UdsAddressUniqueIdPB::BACKEND_ID) { |
| 1129 | ClusterMembershipMgr::SnapshotPtr membership_snapshot = |
| 1130 | ExecEnv::GetInstance()->cluster_membership_mgr()->GetSnapshot(); |
| 1131 | DCHECK(membership_snapshot.get() != nullptr); |
| 1132 | for (const auto& it : membership_snapshot->current_backends) { |
| 1133 | // Compare resolved IP addresses and ports. |
| 1134 | if (it.second.ip_address() == ip_address && it.second.address().port() == port) { |
| 1135 | DCHECK(it.second.has_backend_id()); |
| 1136 | backend_id = it.second.backend_id(); |
| 1137 | break; |
| 1138 | } |
| 1139 | } |
| 1140 | } |
| 1141 | } |
| 1142 | string krpc_error = "RemoteShutdown() RPC failed: Network error"; |
| 1143 | string krpc_error2 = "RemoteShutdown() RPC failed: Timed out"; |
| 1144 | NetworkAddressPB krpc_addr = MakeNetworkAddressPB(ip_address, port, backend_id, |
| 1145 | ExecEnv::GetInstance()->rpc_mgr()->GetUdsAddressUniqueId()); |
| 1146 | std::unique_ptr<ControlServiceProxy> proxy; |
| 1147 | Status get_proxy_status = |
| 1148 | ControlService::GetProxy(krpc_addr, request.backend.hostname, &proxy); |
| 1149 | if (!get_proxy_status.ok()) { |
| 1150 | return Status( |
| 1151 | Substitute("Could not get Proxy to ControlService at $0 with error: $1.", |
nothing calls this directly
no test coverage detected