telemetry_pgstat.c
text/x-csrc
Filename: telemetry_pgstat.c
Type: text/x-csrc
Part: 1
#include "postgres.h"
#include <unistd.h>
#include "access/transam.h"
#include "access/xact.h"
#include "lib/dshash.h"
#include "pgstat.h"
#include "port/atomics.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lwlock.h"
#include "storage/pg_shmem.h"
#include "storage/shmem.h"
#include "utils/guc.h"
#include "utils/memutils.h"
#include "utils/pgstat_internal.h"
#include "utils/timestamp.h"
#include <math.h>
#include <float.h>
/* SQL callable interface to return metrics */
Datum
pg_stat_dump_stats(PG_FUNCTION_ARGS);
/* C-level call for telemetry monitoring agent to call */
char *
pg_stat_dump_stats_string(void);
/* XXX */
typedef void (*dump_stats_cb)(PgStat_Kind kind, const char *kind_name, Oid dboid, Oid objoid, const char *name, void *body, void *closure);
static void
pgstat_dump_stats(dump_stats_cb callback, void *closure)
{
PgStat_Kind kind;
dshash_seq_status hstat;
PgStatShared_HashEntry *ps;
struct timespec tstart={0,0}, tend={0,0};
clock_gettime(CLOCK_MONOTONIC, &tstart);
pgstat_assert_is_up();
pgstat_fetch_consistency = PGSTAT_FETCH_CONSISTENCY_NONE;
elog(NOTICE, "dumping stats contents");
/*
* XXX: The following could now be generalized to just iterate over
* pgstat_kind_infos instead of knowing about the different kinds of
* stats.
*/
for (kind=PGSTAT_KIND_FIRST_VALID; kind <= PGSTAT_KIND_LAST; kind++)
{
if (pgstat_get_kind_info(kind)->fixed_amount)
pgstat_snapshot_fixed(kind);
}
callback(PGSTAT_KIND_ARCHIVER, pgstat_get_kind_info(PGSTAT_KIND_ARCHIVER)->name, InvalidOid, InvalidOid, NULL, &pgStatLocal.snapshot.archiver, closure);
callback(PGSTAT_KIND_BGWRITER, pgstat_get_kind_info(PGSTAT_KIND_BGWRITER)->name, InvalidOid, InvalidOid, NULL, &pgStatLocal.snapshot.bgwriter, closure);
callback(PGSTAT_KIND_CHECKPOINTER, pgstat_get_kind_info(PGSTAT_KIND_CHECKPOINTER)->name, InvalidOid, InvalidOid, NULL, &pgStatLocal.snapshot.checkpointer, closure);
for (int i = 0 ; i < SLRU_NUM_ELEMENTS; i++ ) {
callback(PGSTAT_KIND_SLRU, pgstat_get_kind_info(PGSTAT_KIND_SLRU)->name, InvalidOid, InvalidOid, slru_names[i], &pgStatLocal.snapshot.slru[i], closure);
}
callback(PGSTAT_KIND_WAL, pgstat_get_kind_info(PGSTAT_KIND_WAL)->name, InvalidOid, InvalidOid, NULL, &pgStatLocal.snapshot.wal, closure);
/*
* Walk through the stats entries
*/
dshash_seq_init(&hstat, pgStatLocal.shared_hash, false);
while ((ps = dshash_seq_next(&hstat)) != NULL)
{
PgStatShared_Common *shstats;
void *body;
const PgStat_KindInfo *kind_info = NULL;
CHECK_FOR_INTERRUPTS();
/* we may have some "dropped" entries not yet removed, skip them */
Assert(!ps->dropped);
if (ps->dropped)
continue;
shstats = (PgStatShared_Common *) dsa_get_address(pgStatLocal.dsa, ps->body);
body = pgstat_get_entry_data(ps->key.kind, shstats);
kind_info = pgstat_get_kind_info(ps->key.kind);
/* if not dropped the valid-entry refcount should exist */
Assert(pg_atomic_read_u32(&ps->refcount) > 0);
if (!kind_info->to_serialized_name)
{
callback(ps->key.kind, kind_info->name, ps->key.dboid, ps->key.objoid, NULL, body, closure);
}
else
{
/* stats entry identified by name on disk (e.g. replication slots) */
NameData name;
kind_info->to_serialized_name(shstats, &name);
callback(ps->key.kind, kind_info->name, InvalidOid, InvalidOid, NameStr(name), body, closure);
}
}
dshash_seq_term(&hstat);
clock_gettime(CLOCK_MONOTONIC, &tend);
elog(NOTICE,
"pgstat_dump_stats took about %0.3f ms\n",
1000 * (((double)tend.tv_sec + 1.0e-9*tend.tv_nsec) -
((double)tstart.tv_sec + 1.0e-9*tstart.tv_nsec))
);
}
/*
* Lots of work left to do here
*
* Prometheus expects double precision floats and all our counters are
* integers so the best strategy is just to have them wrap at 2^53 to avoid
* losing precision. Prometheus treats that as a reset so we trade absolute
* value for precision but since they're counters that's fine.
*
* Our timestamps are 64-bit microseconds and Prometheus standard is to use
* double precision in units of seconds. That loses precision for dates far in
* the future and past but these stats just have things like the time of the
* last failure.
*
*/
#define COUNTER_MASK ((1LL<<DBL_MANT_DIG)-1)
static void emit_counter(const char *name, char *labels, PgStat_Counter counter, void *closure)
{
StringInfo response = closure;
int64 val = counter & COUNTER_MASK;
/* Assert check that we are printed values that don't lose precision in
* doubles */
double valf = (double)val;
Assert((int64)valf == val);
if (labels)
appendStringInfo(response, "%s{%s}=%lu\n", name, labels, val);
else
appendStringInfo(response, "%s=%lu\n", name, val);
}
static void emit_timestamptz(const char *name, char *labels, TimestampTz timestamp, void *closure)
{
StringInfo response = closure;
double val = timestamp== 0 ? NAN : (double)timestamp / 1000000 + SECS_PER_DAY*(POSTGRES_EPOCH_JDATE-UNIX_EPOCH_JDATE);
if (labels)
appendStringInfo(response, "%s{%s}=%lf\n", name, labels, val);
else
appendStringInfo(response, "%s=%lf\n", name, val);
}
static void
dump_stats_callback(PgStat_Kind kind, const char *kind_name, Oid dboid, Oid objoid, const char *name, void *body, void *closure) {
if (dboid == InvalidOid && objoid == InvalidOid && name == NULL) {
elog(WARNING, "stats for \"%s\" with no key?!", kind_name);
}
switch(kind) {
/* stats for variable-numbered objects */
case PGSTAT_KIND_DATABASE:
{
PgStat_StatDBEntry *stats = body;
/* db=0 represents the background workers apparently? */
char *labels = dboid==0 ? NULL : psprintf("db=\"%d\"", dboid);
emit_counter("xact_commit", labels, stats->n_xact_commit, closure);
emit_counter("xact_rollback", labels, stats->n_xact_rollback, closure);
emit_counter("blks_fetched", labels, stats->n_blocks_fetched, closure);
emit_counter("blks_hit", labels, stats->n_blocks_hit, closure);
emit_counter("tup_returned", labels, stats->n_tuples_returned, closure);
emit_counter("tup_fetched", labels, stats->n_tuples_fetched, closure);
emit_counter("tup_inserted", labels, stats->n_tuples_inserted, closure);
emit_counter("tup_updated", labels, stats->n_tuples_updated, closure);
emit_counter("tup_deleted", labels, stats->n_tuples_deleted, closure);
emit_timestamptz("last_autovac_time", labels, stats->last_autovac_time, closure);
/* Why do we do this?? */
emit_counter("conflicts",
labels,
(stats->n_conflict_tablespace +
stats->n_conflict_lock +
stats->n_conflict_snapshot +
stats->n_conflict_bufferpin +
stats->n_conflict_startup_deadlock),
closure);
/*
emit_counter("conflicts_tablespace", labels, stats->n_conflict_tablespace, closure);
emit_counter("conflicts_lock", labels, stats->n_conflict_lock, closure);
emit_counter("conflicts_snapshot", labels, stats->n_conflict_snapshot, closure);
emit_counter("conflicts_bufferpin", labels, stats->n_conflict_bufferpin, closure);
emit_counter("conflicts_startup_deadlock", labels, stats->n_conflict_startup_deadlock, closure);
*/
emit_counter("temp_files", labels, stats->n_temp_files, closure);
emit_counter("temp_bytes", labels, stats->n_temp_bytes, closure);
emit_counter("deadlocks", labels, stats->n_deadlocks, closure);
emit_counter("checksum_failures", labels, stats->n_checksum_failures, closure);
emit_timestamptz("last_checksum_failure", labels, stats->last_checksum_failure, closure);
emit_counter("blk_read_time", labels, stats->n_block_read_time, closure);
emit_counter("blk_write_time", labels, stats->n_block_write_time, closure);
emit_counter("numbackends", labels, stats->n_sessions, closure);
emit_counter("session_time", labels, stats->total_session_time, closure);
emit_counter("active_time", labels, stats->total_active_time, closure);
emit_counter("idle_in_transaction_time", labels, stats->total_idle_in_xact_time, closure);
emit_counter("sessions_abandoned", labels, stats->n_sessions_abandoned, closure);
emit_counter("sessions_fatal", labels, stats->n_sessions_fatal, closure);
emit_counter("sessions_killed", labels, stats->n_sessions_killed, closure);
/*emit_timestamptz("stat_reset_timestamp", labels, stats->stat_reset_timestamp, closure);*/
break;
}
case PGSTAT_KIND_RELATION:
{
PgStat_StatTabEntry *stats = body;
/* db=0 represents the shared tables apparently? */
char *labels = dboid==0
? psprintf("oid=\"%d\"", objoid)
: psprintf("db=\"%d\",oid=\"%d\"", dboid,objoid);
emit_counter("numscans", labels, stats->numscans, closure);
emit_counter("tuples_returned", labels, stats->tuples_returned, closure);
emit_counter("tuples_fetched", labels, stats->tuples_fetched, closure);
emit_counter("tuples_inserted", labels, stats->tuples_inserted, closure);
emit_counter("tuples_updated", labels, stats->tuples_updated, closure);
emit_counter("tuples_deleted", labels, stats->tuples_deleted, closure);
emit_counter("tuples_hot_updated", labels, stats->tuples_hot_updated, closure);
emit_counter("n_live_tuples", labels, stats->n_live_tuples, closure);
emit_counter("n_dead_tuples", labels, stats->n_dead_tuples, closure);
emit_counter("changes_since_analyze", labels, stats->changes_since_analyze, closure);
emit_counter("inserts_since_vacuum", labels, stats->inserts_since_vacuum, closure);
emit_counter("blocks_fetched", labels, stats->blocks_fetched, closure);
emit_counter("blocks_hit", labels, stats->blocks_hit, closure);
emit_timestamptz("vacuum_timestamp", labels, stats->vacuum_timestamp, closure);
emit_counter("vacuum_count", labels, stats->vacuum_count, closure);
emit_timestamptz("autovac_vacuum_timestamp", labels, stats->autovac_vacuum_timestamp, closure);
emit_counter("autovac_vacuum_count", labels, stats->autovac_vacuum_count, closure);
emit_timestamptz("analyze_timestamp", labels, stats->analyze_timestamp, closure);
emit_counter("analyze_count", labels, stats->analyze_count, closure);
emit_timestamptz("autovac_analyze_timestamp", labels, stats->autovac_analyze_timestamp, closure);
emit_counter("autovac_analyze_count", labels, stats->autovac_analyze_count, closure);
break;
}
case PGSTAT_KIND_FUNCTION:
{
PgStat_StatFuncEntry *stats = body;
char *labels = psprintf("db=\"%d\",oid=\"%d\"", dboid,objoid);
emit_counter("calls", labels, stats->f_numcalls, closure);
emit_counter("total_time", labels, stats->f_total_time, closure);
emit_counter("self_time", labels, stats->f_self_time, closure);
break;
}
case PGSTAT_KIND_REPLSLOT:
{
PgStat_StatReplSlotEntry *stats = body;
char *labels = psprintf("slot_name=\"%s\"", name);
emit_counter("spill_txns", labels, stats->spill_txns, closure);
emit_counter("spill_count", labels, stats->spill_count, closure);
emit_counter("spill_bytes", labels, stats->spill_bytes, closure);
emit_counter("stream_txns", labels, stats->stream_txns, closure);
emit_counter("stream_count", labels, stats->stream_count, closure);
emit_counter("stream_bytes", labels, stats->stream_bytes, closure);
emit_counter("total_txns", labels, stats->total_txns, closure);
emit_counter("total_bytes", labels, stats->total_bytes, closure);
/*emit_timestamptz("stat_reset_timestamp", labels, stats->stat_reset_timestamp, closure);*/
break;
}
case PGSTAT_KIND_SUBSCRIPTION:
{
PgStat_StatSubEntry *stats = body;
char *labels = psprintf("db=\"%d\",oid=\"%d\"", dboid,objoid);
emit_counter("apply_error_count", labels, stats->apply_error_count, closure);
emit_counter("sync_error_count", labels, stats->sync_error_count, closure);
/*emit_timestamptz("stat_reset_timestamp", labels, stats->stat_reset_timestamp, closure);*/
break;
}
/* stats for fixed-numbered objects -- these are actually invoked
* explicitly from dump_stats on the snapshot objects rather than through
* the dshash iteration */
case PGSTAT_KIND_ARCHIVER:
{
PgStat_ArchiverStats *stats = body;
char *labels = NULL;
emit_counter("archived_count", labels, stats->archived_count, closure);
/* last_archived_wal*/
emit_timestamptz("last_archived_timestamp", labels, stats->last_archived_timestamp, closure);
emit_counter("failed_count", labels, stats->failed_count, closure);
/* last_failed_wal */
emit_timestamptz("last_failed_timestamp", labels, stats->last_failed_timestamp, closure);
/*emit_timestamptz("stat_reset_timestamp", labels, stats->stat_reset_timestamp, closure);*/
break;
}
case PGSTAT_KIND_BGWRITER:
{
PgStat_BgWriterStats *stats = body;
char *labels = NULL;
emit_counter("buffers_clean", labels, stats->buf_written_clean, closure);
emit_counter("maxwritten_clean", labels, stats->maxwritten_clean, closure);
emit_counter("buffers_alloc", labels, stats->buf_alloc, closure);
/*emit_timestamptz("stat_reset_timestamp", labels, stats->stat_reset_timestamp, closure);*/
break;
}
case PGSTAT_KIND_CHECKPOINTER:
{
PgStat_CheckpointerStats *stats = body;
char *labels = NULL;
emit_counter("checkpoints_timed", labels, stats->timed_checkpoints, closure);
emit_counter("checkpoints_req", labels, stats->requested_checkpoints, closure);
emit_counter("checkpoint_write_time", labels, stats->checkpoint_write_time, closure);
emit_counter("checkpoint_sync_time", labels, stats->checkpoint_sync_time, closure);
emit_counter("buffers_checkpoint", labels, stats->buf_written_checkpoints, closure);
emit_counter("buffers_backend", labels, stats->buf_written_backend, closure);
emit_counter("buffers_backend_fsync", labels, stats->buf_fsync_backend, closure);
break;
}
case PGSTAT_KIND_SLRU:
{
PgStat_SLRUStats *stats = body;
char *labels = psprintf("slru=\"%s\"", name);
emit_counter("blks_zeroed", labels, stats->blocks_zeroed, closure);
emit_counter("blks_hit", labels, stats->blocks_hit, closure);
emit_counter("blks_read", labels, stats->blocks_read, closure);
emit_counter("blks_written", labels, stats->blocks_written, closure);
emit_counter("blks_exists", labels, stats->blocks_exists, closure);
emit_counter("flushes", labels, stats->flush, closure);
emit_counter("truncates", labels, stats->truncate, closure);
/*emit_timestamptz("stat_reset_timestamp", labels, stats->stat_reset_timestamp, closure);*/
break;
}
case PGSTAT_KIND_WAL:
{
PgStat_WalStats *stats = body;
char *labels=NULL;
emit_counter("wal_records", labels, stats->wal_records, closure);
emit_counter("wal_fpi", labels, stats->wal_fpi, closure);
emit_counter("wal_bytes", labels, stats->wal_bytes, closure);
emit_counter("wal_buffers_full", labels, stats->wal_buffers_full, closure);
emit_counter("wal_write", labels, stats->wal_write, closure);
emit_counter("wal_sync", labels, stats->wal_sync, closure);
emit_counter("wal_write_time", labels, stats->wal_write_time, closure);
emit_counter("wal_sync_time", labels, stats->wal_sync_time, closure);
/*emit_timestamptz("stat_reset_timestamp", labels, stats->stat_reset_timestamp, closure);*/
break;
}
case PGSTAT_KIND_INVALID:
elog(WARNING, "Invalid stats object type found");
break;
}
}
PG_FUNCTION_INFO_V1(pg_stat_dump_stats);
Datum
pg_stat_dump_stats(PG_FUNCTION_ARGS)
{
StringInfo response = makeStringInfo();
pgstat_dump_stats(&dump_stats_callback, response);
/* Test overflow behaviour, the emitted values should be distinct and wrap around to 0 */
emit_counter("test_value", "value=0", 0, response);
emit_counter("test_value", "value=1<<53-2", (1LL<<53)-2, response);
emit_counter("test_value", "value=1<<53-1", (1LL<<53)-1, response);
emit_counter("test_value", "value=1<<53", 1LL<<53, response);
elog(WARNING, "stats:\n%s", response->data);
elog(WARNING, "DBL_MANT_DIG=%d COUNTER_MASK=%llx", DBL_MANT_DIG, COUNTER_MASK);
PG_RETURN_INT32(response->len);
}
char *
pg_stat_dump_stats_string(void)
{
StringInfo response = makeStringInfo();
pgstat_dump_stats(&dump_stats_callback, response);
return response->data;
}