Function execute_promsql_query

src/environmentd/src/http/sql.rs:116–222

(
    client: &mut AuthedClient,
    query: &PrometheusSqlQuery<'_>,
    metrics_registry: &MetricsRegistry,
    metrics_by_name: &mut BTreeMap<String, GenericGaugeVec<AtomicF64>>,
    cluster: Option

static PER_REPLICA_LABELS: &[&str] = &["replica_full_name", "instance_id", "replica_id"];

async fn execute_promsql_query(
    client: &mut AuthedClient,
    query: &PrometheusSqlQuery<'_>,
    metrics_registry: &MetricsRegistry,
    metrics_by_name: &mut BTreeMap<String, GenericGaugeVec<AtomicF64>>,
    cluster: Option<(&Cluster, &ClusterReplica)>,
) {
    assert_eq!(query.per_replica, cluster.is_some());

    let mut res = SqlResponse {
        results: Vec::new(),
    };

    execute_request(client, query.to_sql_request(cluster), &mut res)
        .await
        .expect("valid SQL query");

    let result = match res.results.as_slice() {
        // Each query issued is preceded by several SET commands
        // to make sure it is routed to the right cluster replica.
        [
            SqlResult::Ok { .. },
            SqlResult::Ok { .. },
            SqlResult::Ok { .. },
            result,
        ] => result,
        // Transient errors are fine, like if the cluster or replica
        // was dropped before the promsql query was executed. We
        // should not see errors in the steady state.
        _ => {
            info!(
                "error executing prometheus query {}: {:?}",
                query.metric_name, res
            );
            return;
        }
    };

    let SqlResult::Rows { desc, rows, .. } = result else {
        info!(
            "did not receive rows for SQL query for prometheus metric {}: {:?}, {:?}",
            query.metric_name, result, cluster
        );
        return;
    };

    let gauge_vec = metrics_by_name
        .entry(query.metric_name.to_string())
        .or_insert_with(|| {
            let mut label_names: Vec<String> = desc
                .columns
                .iter()
                .filter(|col| col.name != query.value_column_name)
                .map(|col| col.name.clone())
                .collect();

            if query.per_replica {
                label_names.extend(PER_REPLICA_LABELS.iter().map(|label| label.to_string()));