// Copyright 2023 PingCAP, Inc.

use lazy_static::lazy_static;
use prometheus::*;
use prometheus_static_metric::*;

make_static_metric! {
    pub label_enum RemoteCompactFailureType {
        queue_full,
        wait_timeout,
    }

    pub label_enum RemoteCopFailureType {
        queue_full,
        wait_timeout,
        black_list,
        memory_limiter,
        snapshot_error,
        snapshot_timeout,
        prefetch_error,
        prefetch_timeout,
        process_timeout,
        process_error,
        delegate_error,
        response_encode_error,
        response_send_error,
    }

    pub label_enum RemoteCopRequestOutcome {
        success,
        failure,
    }

    pub label_enum RemoteCopScanKind {
        processed,
        total,
    }

    pub struct RemoteCompactFailedRequestsCounterVec: IntCounter {
        "type" => RemoteCompactFailureType,
    }

    pub struct RemoteCopFailedRequestsCounterVec: IntCounter {
        "type" => RemoteCopFailureType,
    }

    pub struct RemoteCopRequestsCounterVec: IntCounter {
        "request_type" => {
            dag,
            db9,
            analyze,
            checksum,
            unknown,
        },
        "outcome" => RemoteCopRequestOutcome,
    }

    pub struct RemoteCopScanKeysCounterVec: IntCounter {
        "request_kind" => {
            select,
            index,
            db9_select,
            db9_index,
            analyze_table,
            analyze_index,
            analyze_full_sampling,
            checksum_table,
            checksum_index,
        },
        "kind" => RemoteCopScanKind,
    }

    pub struct RemoteCopScanBytesCounterVec: IntCounter {
        "kind" => RemoteCopScanKind,
    }
}

lazy_static! {
    pub static ref NATIVE_BR_HISTOGRAM_VEC: HistogramVec = register_histogram_vec!(
        "tikv_worker_native_br_duration_seconds",
        "Bucketed histogram of native br duration",
        &["type"],
        exponential_buckets(1.0, 1.5, 20).unwrap() // 1s ~ 0.9h
    )
    .unwrap();

    pub static ref NATIVE_BR_COUNTER_VEC: IntCounterVec = register_int_counter_vec!(
        "tikv_worker_native_br_counter",
        "The counter of native br operations",
        &["type"],
    )
    .unwrap();


    pub static ref REMOTE_COPR_PROCESSING_REQ_COUNTER: IntGauge = register_int_gauge!(
        "tikv_worker_remote_cop_processing_requests_counter",
        "Number of remote coprocessor requests under processing"
    ).unwrap();

    pub static ref REMOTE_COPR_FAILED_REQUESTS_COUNTER_VEC: RemoteCopFailedRequestsCounterVec = register_static_int_counter_vec!(
        RemoteCopFailedRequestsCounterVec,
        "tikv_worker_remote_cop_failed_requests_counter",
        "Number of remote coprocessor requests that failed, by failure reason",
        &["type"]
    )
    .unwrap();

    pub static ref REMOTE_COPR_REQUESTS_COUNTER_VEC: RemoteCopRequestsCounterVec = register_static_int_counter_vec!(
        RemoteCopRequestsCounterVec,
        "tikv_worker_remote_cop_requests_total",
        "Number of terminal remote coprocessor request outcomes by request type",
        &["request_type", "outcome"]
    )
    .unwrap();

    pub static ref REMOTE_COPR_SCAN_KEYS_COUNTER_VEC: RemoteCopScanKeysCounterVec = register_static_int_counter_vec!(
        RemoteCopScanKeysCounterVec,
        "tikv_worker_remote_cop_scan_keys_total",
        "Number of keys scanned by completed remote coprocessor requests, by request kind",
        &["request_kind", "kind"]
    )
    .unwrap();

    pub static ref REMOTE_COPR_SCAN_BYTES_COUNTER_VEC: RemoteCopScanBytesCounterVec = register_static_int_counter_vec!(
        RemoteCopScanBytesCounterVec,
        "tikv_worker_remote_cop_scan_bytes_total",
        "Number of undiscounted bytes scanned by completed remote coprocessor requests",
        &["kind"]
    )
    .unwrap();

    pub static ref REMOTE_COPR_RANGE_COUNT_HISTOGRAM: Histogram = register_histogram!(
        "tikv_worker_remote_cop_request_ranges",
        "Number of key ranges in a remote coprocessor request",
        exponential_buckets(1.0, 2.0, 20).unwrap()
    )
    .unwrap();
    pub static ref REMOTE_COPR_DAG_REQ_COUNTER: IntCounter = register_int_counter!(
        "tikv_worker_remote_cop_dag_request_counter",
        "Total count of remote copr requests",
    )
    .unwrap();

    pub static ref REMOTE_COPR_TOPN_PAGING_REWRITE_TOTAL: IntCounter = register_int_counter!(
        "tikv_worker_remote_cop_topn_paging_rewrite_total",
        "Number of remote TopN requests whose paging size is rewritten by the worker",
    )
    .unwrap();

    pub static ref REMOTE_COPR_DAG_RESP_SIZE: IntCounter = register_int_counter!(
        "tikv_worker_remote_cop_dag_response_size",
        "Total size of remote copr responses",
    )
    .unwrap();

    pub static ref REMOTE_ANALYZE_REQ_COUNTER: IntCounter = register_int_counter!(
        "tikv_worker_remote_analyze_request_counter",
        "Total count of remote analyze requests",
    )
    .unwrap();

    pub static ref REMOTE_ANALYZE_RESP_SIZE: IntCounter = register_int_counter!(
        "tikv_worker_remote_analyze_response_size",
        "Total size of remote analyze responses",
    )
    .unwrap();

    pub static ref REMOTE_CHECKSUM_REQ_COUNTER: IntCounter = register_int_counter!(
        "tikv_worker_remote_checksum_request_counter",
        "Total count of remote checksum requests",
    )
    .unwrap();

    pub static ref REMOTE_CHECKSUM_RESP_SIZE: IntCounter = register_int_counter!(
        "tikv_worker_remote_checksum_response_size",
        "Total size of remote checksum responses",
    )
    .unwrap();

    pub static ref REMOTE_COPR_SNAPSHOT_HISTOGRAM: Histogram = register_histogram!(
        "tikv_worker_remote_cop_snapshot_duration_seconds",
        "Bucketed histogram of remote copr snapshot duration",
        exponential_buckets(0.0005, 2.0, 20).unwrap()
    ).unwrap();

    pub static ref REMOTE_COPR_PREFETCH_HISTOGRAM: Histogram = register_histogram!(
        "tikv_worker_remote_cop_prefetch_duration_seconds",
        "Bucketed histogram of remote copr prefetch segments duration",
        exponential_buckets(0.0005, 2.0, 20).unwrap()
    ).unwrap();

    pub static ref REMOTE_COPR_PREFETCH_CACHE_HIT_PERCENT_HISTOGRAM: Histogram = register_histogram!(
        "tikv_worker_remote_cop_prefetch_cache_hit_percent",
        "Bucketed histogram of cache hit percent for remote copr prefetch segments",
        linear_buckets(50.0, 2.0, 25).unwrap()
    )
    .unwrap();

    pub static ref REMOTE_COPR_PROCESS_HISTOGRAM: Histogram = register_histogram!(
        "tikv_worker_remote_cop_process_duration_seconds",
        "Bucketed histogram of remote copr process duration",
        exponential_buckets(0.0005, 2.0, 20).unwrap()
    ).unwrap();

    pub static ref REMOTE_COPR_REQ_HANDLE_HISTOGRAM: Histogram = register_histogram!(
        "tikv_worker_remote_cop_request_duration_seconds",
        "Bucketed histogram of remote copr request duration",
        exponential_buckets(0.0005, 2.0, 20).unwrap()
    ).unwrap();

    pub static ref REMOTE_COMPACT_REQ_HANDLE_HISTOGRAM: Histogram = register_histogram!(
        "tikv_worker_remote_compact_request_duration_seconds",
        "Bucketed histogram of remote compaction request duration",
        exponential_buckets(0.0005, 2.0, 20).unwrap()
    ).unwrap();

    pub static ref REMOTE_COMPACT_PROCESSING_REQUESTS_COUNTER: IntGauge = register_int_gauge!(
        "tikv_worker_remote_compact_processing_requests_counter",
        "Number of remote compaction requests being processed"
    ).unwrap();

    pub static ref WORKER_LIMITER_REQUEST_WAIT_HISTOGRAM: HistogramVec = register_histogram_vec!(
        "tikv_worker_limiter_request_wait_duration_seconds",
        "Bucketed histogram of worker limiter request wait duration",
        & ["type"],
        exponential_buckets(0.0005, 2.0, 20).unwrap()
    ).unwrap();

    pub static ref WORKER_LIMITER_WAITING_REQUESTS_COUNTER_VEC: IntGaugeVec = register_int_gauge_vec!(
        "tikv_worker_limiter_waiting_requests_counter",
        "The number of requests waiting for permits",
        &["type"]
    )
    .unwrap();

    pub static ref REMOTE_COMPACT_FAILED_REQUESTS_COUNTER_VEC: RemoteCompactFailedRequestsCounterVec = register_static_int_counter_vec!(
        RemoteCompactFailedRequestsCounterVec,
        "tikv_worker_remote_compact_failed_requests_counter",
        "Number of remote compaction requests that failed to acquire permit, by failure reason",
        &["type"]
    )
    .unwrap();

    pub static ref WORKER_SCALER_QUERY_FAILURES_COUNTER_VEC: IntCounterVec = register_int_counter_vec!(
        "tikv_worker_worker_scaler_query_failures_counter",
        "Total count of worker scaler failures in querying task state",
        &["task_id"],
    )
    .unwrap();

    pub static ref WORKER_MEMORY_LIMITER_CURRENT_USED: IntGauge = register_int_gauge!(
        "tikv_worker_memory_limiter_current_used",
        "Current used memory reported by worker memory limiter",
    ).unwrap();

    pub static ref SCHEMA_MANAGER_SYNC_LOOP_COUNT: IntCounter = register_int_counter!(
        "tikv_worker_schema_manager_sync_loop_count",
        "Total count of schema manager sync loop",
    ).unwrap();

    pub static ref SCHEMA_MANAGER_SYNC_LOOP_ERROR_COUNT: IntCounterVec = register_int_counter_vec!(
        "tikv_worker_schema_manager_sync_loop_error_count",
        "Total count of schema manager sync loop errors",
        &["type"],
    ).unwrap();
    pub static ref NATIVE_BR_V1X_COUNTER_VEC: IntCounterVec = register_int_counter_vec!(
        "tikv_worker_native_br_v1x_counter",
        "The counter of native br v1x operations",
        &["type", "result"]
    )
    .unwrap();

    pub static ref NATIVE_BR_V1X_BACKGROUND_TASKS: IntGaugeVec = register_int_gauge_vec!(
        "tikv_worker_native_running_tasks",
        "The running tasks of v1x API",
        &["type"]
    ).unwrap();
}

#[cfg(test)]
mod tests {
    use prometheus::core::Collector;

    use super::*;

    #[test]
    fn test_remote_cop_request_metric_has_only_terminal_outcomes() {
        let vector = IntCounterVec::new(
            Opts::new(
                "test_remote_cop_requests_total",
                "Remote cop request outcomes for this test",
            ),
            &["request_type", "outcome"],
        )
        .unwrap();
        let _static_metrics = RemoteCopRequestsCounterVec::from(&vector);

        let mut outcomes = vector
            .collect()
            .into_iter()
            .flat_map(|family| family.get_metric().to_vec())
            .flat_map(|metric| metric.get_label().to_vec())
            .filter(|label| label.get_name() == "outcome")
            .map(|label| label.get_value().to_owned())
            .collect::<Vec<_>>();
        outcomes.sort();
        outcomes.dedup();

        assert_eq!(outcomes, ["failure", "success"]);
    }

    #[test]
    fn test_remote_compact_failure_metric_has_only_compaction_labels() {
        let vector = IntCounterVec::new(
            Opts::new(
                "test_remote_compact_failed_requests_total",
                "Remote compaction failures for this test",
            ),
            &["type"],
        )
        .unwrap();
        let _static_metrics = RemoteCompactFailedRequestsCounterVec::from(&vector);

        let mut failure_types = vector
            .collect()
            .into_iter()
            .flat_map(|family| family.get_metric().to_vec())
            .flat_map(|metric| metric.get_label().to_vec())
            .filter(|label| label.get_name() == "type")
            .map(|label| label.get_value().to_owned())
            .collect::<Vec<_>>();
        failure_types.sort();

        assert_eq!(failure_types, ["queue_full", "wait_timeout"]);
    }
}