TODO: do we need host_id_ to come from host_addr or can it just take the same id the Scheduler has (coming from the StatestoreSubscriber)?
| 779 | // TODO: do we need host_id_ to come from host_addr or can it just take the same id |
| 780 | // the Scheduler has (coming from the StatestoreSubscriber)? |
| 781 | AdmissionController::AdmissionController(ClusterMembershipMgr* cluster_membership_mgr, |
| 782 | StatestoreSubscriber* subscriber, RequestPoolService* request_pool_service, |
| 783 | MetricGroup* metrics, Scheduler* scheduler, PoolMemTrackerRegistry* pool_mem_trackers, |
| 784 | const TNetworkAddress& host_addr) |
| 785 | : cluster_membership_mgr_(cluster_membership_mgr), |
| 786 | subscriber_(subscriber), |
| 787 | request_pool_service_(request_pool_service), |
| 788 | metrics_group_(metrics->GetOrCreateChildGroup("admission-controller")), |
| 789 | scheduler_(scheduler), |
| 790 | pool_mem_trackers_(pool_mem_trackers), |
| 791 | host_id_(TNetworkAddressToString(host_addr)), |
| 792 | thrift_serializer_(false), |
| 793 | done_(false) { |
| 794 | cluster_membership_mgr_->RegisterUpdateCallbackFn( |
| 795 | [this](const ClusterMembershipMgr::SnapshotPtr& snapshot) { |
| 796 | this->UpdateExecGroupMetricMap(snapshot); |
| 797 | }); |
| 798 | total_dequeue_failed_coordinator_limited_ = |
| 799 | metrics_group_->AddCounter(TOTAL_DEQUEUE_FAILED_COORDINATOR_LIMITED, 0); |
| 800 | compressed_size_metric_ = metrics_group_->RegisterMetric(new HistogramMetric( |
| 801 | MetricDefs::Get(EXEC_REQ_COMPRESSED_SIZE_KEY), numeric_limits<int64_t>::max(), 3)); |
| 802 | uncompressed_size_metric_ = metrics_group_->RegisterMetric( |
| 803 | new HistogramMetric(MetricDefs::Get(EXEC_REQ_UNCOMPRESSED_SIZE_KEY), |
| 804 | numeric_limits<int64_t>::max(), 3)); |
| 805 | compression_ratio_metric_ = metrics_group_->RegisterMetric( |
| 806 | new HistogramMetric(MetricDefs::Get(EXEC_REQ_COMPRESSION_RATIO_KEY), 10000, 3)); |
| 807 | if (FLAGS_cluster_membership_topic_id.empty()) { |
| 808 | request_queue_topic_name_ = Statestore::IMPALA_REQUEST_QUEUE_TOPIC; |
| 809 | } else { |
| 810 | request_queue_topic_name_ = |
| 811 | FLAGS_cluster_membership_topic_id + '-' + Statestore::IMPALA_REQUEST_QUEUE_TOPIC; |
| 812 | } |
| 813 | } |
| 814 | |
| 815 | AdmissionController::~AdmissionController() { |
| 816 | // If the dequeue thread is not running (e.g. if Init() fails), then there is |
nothing calls this directly
no test coverage detected