telemetry_pgstat.c

text/x-csrc

Filename: telemetry_pgstat.c
Type: text/x-csrc
Part: 1
Message: Re: shared-memory based stats collector - v70
#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;
}