mirror of
https://github.com/neondatabase/neon.git
synced 2026-01-06 13:02:55 +00:00
[compute_tools] Add /insights endpoint to compute_ctl (#3704)
This commit adds a basic HTTP API endpoint that allows scraping the `pg_stat_statements` data and getting a list of slow queries. New insights like cache hit rate and so on could be added later. Extension `pg_stat_statements` is checked / created only if compute tries to load the corresponding shared library. The latter is configured by control-plane and currently covered with feature flag. Co-authored by Eduard Dyckman (bird.duskpoet@gmail.com)
This commit is contained in:
@@ -25,6 +25,7 @@ use anyhow::{Context, Result};
|
|||||||
use chrono::{DateTime, Utc};
|
use chrono::{DateTime, Utc};
|
||||||
use postgres::{Client, NoTls};
|
use postgres::{Client, NoTls};
|
||||||
use serde::{Serialize, Serializer};
|
use serde::{Serialize, Serializer};
|
||||||
|
use tokio_postgres;
|
||||||
use tracing::{info, instrument, warn};
|
use tracing::{info, instrument, warn};
|
||||||
|
|
||||||
use crate::checker::create_writability_check_data;
|
use crate::checker::create_writability_check_data;
|
||||||
@@ -284,6 +285,7 @@ impl ComputeNode {
|
|||||||
handle_role_deletions(self, &mut client)?;
|
handle_role_deletions(self, &mut client)?;
|
||||||
handle_grants(self, &mut client)?;
|
handle_grants(self, &mut client)?;
|
||||||
create_writability_check_data(&mut client)?;
|
create_writability_check_data(&mut client)?;
|
||||||
|
handle_extensions(&self.spec, &mut client)?;
|
||||||
|
|
||||||
// 'Close' connection
|
// 'Close' connection
|
||||||
drop(client);
|
drop(client);
|
||||||
@@ -400,4 +402,43 @@ impl ComputeNode {
|
|||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Select `pg_stat_statements` data and return it as a stringified JSON
|
||||||
|
pub async fn collect_insights(&self) -> String {
|
||||||
|
let mut result_rows: Vec<String> = Vec::new();
|
||||||
|
let connect_result = tokio_postgres::connect(self.connstr.as_str(), NoTls).await;
|
||||||
|
let (client, connection) = connect_result.unwrap();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
if let Err(e) = connection.await {
|
||||||
|
eprintln!("connection error: {}", e);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
let result = client
|
||||||
|
.simple_query(
|
||||||
|
"SELECT
|
||||||
|
row_to_json(pg_stat_statements)
|
||||||
|
FROM
|
||||||
|
pg_stat_statements
|
||||||
|
WHERE
|
||||||
|
userid != 'cloud_admin'::regrole::oid
|
||||||
|
ORDER BY
|
||||||
|
(mean_exec_time + mean_plan_time) DESC
|
||||||
|
LIMIT 100",
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
if let Ok(raw_rows) = result {
|
||||||
|
for message in raw_rows.iter() {
|
||||||
|
if let postgres::SimpleQueryMessage::Row(row) = message {
|
||||||
|
if let Some(json) = row.get(0) {
|
||||||
|
result_rows.push(json.to_string());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
format!("{{\"pg_stat_statements\": [{}]}}", result_rows.join(","))
|
||||||
|
} else {
|
||||||
|
"{{\"pg_stat_statements\": []}}".to_string()
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -33,6 +33,13 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
|
|||||||
Response::new(Body::from(serde_json::to_string(&compute.metrics).unwrap()))
|
Response::new(Body::from(serde_json::to_string(&compute.metrics).unwrap()))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Collect Postgres current usage insights
|
||||||
|
(&Method::GET, "/insights") => {
|
||||||
|
info!("serving /insights GET request");
|
||||||
|
let insights = compute.collect_insights().await;
|
||||||
|
Response::new(Body::from(insights))
|
||||||
|
}
|
||||||
|
|
||||||
(&Method::POST, "/check_writability") => {
|
(&Method::POST, "/check_writability") => {
|
||||||
info!("serving /check_writability POST request");
|
info!("serving /check_writability POST request");
|
||||||
let res = crate::checker::check_writability(compute).await;
|
let res = crate::checker::check_writability(compute).await;
|
||||||
|
|||||||
@@ -10,12 +10,12 @@ paths:
|
|||||||
/status:
|
/status:
|
||||||
get:
|
get:
|
||||||
tags:
|
tags:
|
||||||
- "info"
|
- Info
|
||||||
summary: Get compute node internal status
|
summary: Get compute node internal status
|
||||||
description: ""
|
description: ""
|
||||||
operationId: getComputeStatus
|
operationId: getComputeStatus
|
||||||
responses:
|
responses:
|
||||||
"200":
|
200:
|
||||||
description: ComputeState
|
description: ComputeState
|
||||||
content:
|
content:
|
||||||
application/json:
|
application/json:
|
||||||
@@ -25,27 +25,43 @@ paths:
|
|||||||
/metrics.json:
|
/metrics.json:
|
||||||
get:
|
get:
|
||||||
tags:
|
tags:
|
||||||
- "info"
|
- Info
|
||||||
summary: Get compute node startup metrics in JSON format
|
summary: Get compute node startup metrics in JSON format
|
||||||
description: ""
|
description: ""
|
||||||
operationId: getComputeMetricsJSON
|
operationId: getComputeMetricsJSON
|
||||||
responses:
|
responses:
|
||||||
"200":
|
200:
|
||||||
description: ComputeMetrics
|
description: ComputeMetrics
|
||||||
content:
|
content:
|
||||||
application/json:
|
application/json:
|
||||||
schema:
|
schema:
|
||||||
$ref: "#/components/schemas/ComputeMetrics"
|
$ref: "#/components/schemas/ComputeMetrics"
|
||||||
|
|
||||||
|
/insights:
|
||||||
|
get:
|
||||||
|
tags:
|
||||||
|
- Info
|
||||||
|
summary: Get current compute insights in JSON format
|
||||||
|
description: |
|
||||||
|
Note, that this doesn't include any historical data
|
||||||
|
operationId: getComputeInsights
|
||||||
|
responses:
|
||||||
|
200:
|
||||||
|
description: Compute insights
|
||||||
|
content:
|
||||||
|
application/json:
|
||||||
|
schema:
|
||||||
|
$ref: "#/components/schemas/ComputeInsights"
|
||||||
|
|
||||||
/check_writability:
|
/check_writability:
|
||||||
post:
|
post:
|
||||||
tags:
|
tags:
|
||||||
- "check"
|
- Check
|
||||||
summary: Check that we can write new data on this compute
|
summary: Check that we can write new data on this compute
|
||||||
description: ""
|
description: ""
|
||||||
operationId: checkComputeWritability
|
operationId: checkComputeWritability
|
||||||
responses:
|
responses:
|
||||||
"200":
|
200:
|
||||||
description: Check result
|
description: Check result
|
||||||
content:
|
content:
|
||||||
text/plain:
|
text/plain:
|
||||||
@@ -96,6 +112,15 @@ components:
|
|||||||
type: string
|
type: string
|
||||||
description: Text of the error during compute startup, if any
|
description: Text of the error during compute startup, if any
|
||||||
|
|
||||||
|
ComputeInsights:
|
||||||
|
type: object
|
||||||
|
properties:
|
||||||
|
pg_stat_statements:
|
||||||
|
description: Contains raw output from pg_stat_statements in JSON format
|
||||||
|
type: array
|
||||||
|
items:
|
||||||
|
type: object
|
||||||
|
|
||||||
ComputeStatus:
|
ComputeStatus:
|
||||||
type: string
|
type: string
|
||||||
enum:
|
enum:
|
||||||
|
|||||||
@@ -63,6 +63,8 @@ impl GenericOption {
|
|||||||
/// Represent `GenericOption` as configuration option.
|
/// Represent `GenericOption` as configuration option.
|
||||||
pub fn to_pg_setting(&self) -> String {
|
pub fn to_pg_setting(&self) -> String {
|
||||||
if let Some(val) = &self.value {
|
if let Some(val) = &self.value {
|
||||||
|
// TODO: check in the console DB that we don't have these settings
|
||||||
|
// set for any non-deleted project and drop this override.
|
||||||
let name = match self.name.as_str() {
|
let name = match self.name.as_str() {
|
||||||
"safekeepers" => "neon.safekeepers",
|
"safekeepers" => "neon.safekeepers",
|
||||||
"wal_acceptor_reconnect" => "neon.safekeeper_reconnect_timeout",
|
"wal_acceptor_reconnect" => "neon.safekeeper_reconnect_timeout",
|
||||||
|
|||||||
@@ -515,3 +515,18 @@ pub fn handle_grants(node: &ComputeNode, client: &mut Client) -> Result<()> {
|
|||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Create required system extensions
|
||||||
|
#[instrument(skip_all)]
|
||||||
|
pub fn handle_extensions(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
|
||||||
|
if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
|
||||||
|
if libs.contains("pg_stat_statements") {
|
||||||
|
// Create extension only if this compute really needs it
|
||||||
|
let query = "CREATE EXTENSION IF NOT EXISTS pg_stat_statements";
|
||||||
|
info!("creating system extensions with query: {}", query);
|
||||||
|
client.simple_query(query)?;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user