telemetry.c
text/x-csrc
Filename: telemetry.c
Type: text/x-csrc
Part: 0
/*-------------------------------------------------------------------------
*
* telemetry.c
*
* Most of this code was copied from pg_prewarm.c as a template.
*
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include <unistd.h>
#include <stdio.h>
#include <stdarg.h>
#include "access/relation.h"
#include "access/xact.h"
#include "catalog/pg_class.h"
#include "catalog/pg_type.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "postmaster/interrupt.h"
#include "storage/buf_internals.h"
#include "storage/dsm.h"
#include "storage/ipc.h"
#include "storage/latch.h"
#include "storage/lwlock.h"
#include "storage/proc.h"
#include "storage/procsignal.h"
#include "storage/shmem.h"
#include "storage/smgr.h"
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/datetime.h"
#include "utils/guc.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/resowner.h"
#include "telemetry.h"
#include "telemetry_pgstat.h"
PG_MODULE_MAGIC;
/* We should already have included what we need to get uint64_t size_t
fd_set socklen_t and struct sockaddr so don't let microhttpd
include extra stuff for them */
#define MHD_PLATFORM_H
#include <microhttpd.h>
/* MHD_USE_EPOLL */
/* MHD_USE_AUTO */
/* MHD_OPTION_CONNECTION_LIMIT */
/* MHD_OPTION_CONNECTION_TIMEOUT */
/* MHD_OPTION_EXTERNAL_LOGGER */
/* Background worker harness */
void _PG_init(void);
/* Actual internal background worker main entry point */
static void telemetry_start_worker(unsigned short port);
/* GUC variables. */
static int telemetry_port = 9187; /* TCP port to listen on for metrics */
static char *telemetry_listen_addresses; /* TCP listen addresses */
static bool telemetry = true; /* start worker automatically on default port */
static enum MHD_Result
telemetry_handler(void *cls,
struct MHD_Connection *connection,
const char *url,
const char *method,
const char *version,
const char *upload_data,
size_t *upload_data_size,
void **con_cls);
/*
* Module load callback.
*/
void
_PG_init(void)
{
DefineCustomIntVariable("telemetry.port",
"TCP Port to serve metrics on by default",
NULL,
&telemetry_port,
9187, 1, 65535,
PGC_SIGHUP,
0, /* flags */
NULL, NULL, NULL /* hooks */
);
DefineCustomStringVariable("telemetry.listen_addresses",
"TCP Listen Addresses to serve metrics on by default",
NULL,
&telemetry_listen_addresses,
"*",
PGC_SIGHUP,
GUC_LIST_INPUT, /* flags */
NULL, NULL, NULL /* hooks */
);
if (!process_shared_preload_libraries_in_progress)
return;
/* can't define PGC_POSTMASTER variable after startup */
DefineCustomBoolVariable("telemetry.start_server",
"Starts the telemetry worker on startup.",
NULL,
&telemetry,
true,
PGC_POSTMASTER,
0,
NULL,
NULL,
NULL);
EmitWarningsOnPlaceholders("telemetry");
/* Register telemetry worker, if enabled. */
if (telemetry)
telemetry_start_worker(telemetry_port);
}
static void telemetry_logger(void *arg, const char *fmt, va_list ap);
struct MHD_OptionItem mhd_ops[] = {
{ MHD_OPTION_EXTERNAL_LOGGER, (intptr_t)telemetry_logger, NULL },
{ MHD_OPTION_CONNECTION_LIMIT, 10, NULL },
{ MHD_OPTION_CONNECTION_TIMEOUT, 30, NULL },
{ MHD_OPTION_END, 0, NULL }
};
/*
* Main entry point for the leader telemetry process. This is invoked
* *in* the background worker process.
*/
void
telemetry_main(Datum main_arg)
{
/* Handle -1 or out of range values? */
unsigned short port = DatumGetInt32(main_arg);
WaitEventSet *waitset;
WaitEvent event;
struct MHD_Daemon *daemon;
enum MHD_Result mhd_retval;
const union MHD_DaemonInfo *mhd_info;
/*fd_set r_fdset, w_fdset, e_fdset;*/
/*int max_fd, i;*/
unsigned long long mhd_timeout;
long timeout;
/* Establish signal handlers; once that's done, unblock signals. */
pqsignal(SIGTERM, SignalHandlerForShutdownRequest);
pqsignal(SIGHUP, SignalHandlerForConfigReload);
pqsignal(SIGUSR1, procsignal_sigusr1_handler);
BackgroundWorkerUnblockSignals();
daemon = MHD_start_daemon(MHD_USE_ERROR_LOG | MHD_USE_EPOLL /* flags */,
port,
NULL /* accept callback */, NULL /* closure */,
telemetry_handler /* default handler */, NULL /* closure */,
MHD_OPTION_ARRAY, mhd_ops,
MHD_OPTION_END);
waitset = CreateWaitEventSet(TopMemoryContext, 3);
mhd_info = MHD_get_daemon_info(daemon, MHD_DAEMON_INFO_EPOLL_FD);
if (mhd_info != NULL && mhd_info->epoll_fd > 0)
{
elog(LOG, "telemetry got %d as epoll file descriptor", mhd_info->epoll_fd);
AddWaitEventToSet(waitset, WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE, mhd_info->epoll_fd, NULL, NULL);
} else {
elog(LOG, "telemetry couldn't get mhd epoll fd");
return;
/*
max_fd = 0;
mhd_retval = MHD_get_fdset(daemon, r_fdset, w_fdset, e_fdset, &max_fd);
for(i=0;i++;i<=max_fd) {
if (FD_ISSET(i, r_fdset)) AddWaitEventToSet(waitset, WL_SOCKET_READABLE, i, NULL, NULL);
if (FD_ISSET(i, w_fdset)) AddWaitEventToSet(waitset, WL_SOCKET_WRITEABLE, i, NULL, NULL);
}
*/
}
AddWaitEventToSet(waitset, WL_LATCH_SET, PGINVALID_SOCKET, MyLatch, NULL);
AddWaitEventToSet(waitset, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET, NULL, NULL);
/* wait for requests */
while (!ShutdownRequestPending)
{
/* In case of a SIGHUP, just reload the configuration. */
if (ConfigReloadPending)
{
int old_port = telemetry_port;
ConfigReloadPending = false;
ProcessConfigFile(PGC_SIGHUP);
if (old_port != telemetry_port) {
;/* restart web server on new port */
}
}
mhd_retval = MHD_run(daemon);
if (mhd_retval != MHD_YES) {
elog(LOG, "mhd_no from MHD_run()");
}
mhd_retval = MHD_get_timeout (daemon, &mhd_timeout);
if (mhd_retval == MHD_YES && mhd_timeout > 0 && mhd_timeout < 10000) {
timeout = (long)timeout;
} else {
timeout = 10000;
}
elog(LOG, "telemetry worker waiting %0.3fs for requests", (double)timeout/1000);
WaitEventSetWait(waitset, (timeout<10000 ? (long)timeout : 10000), &event, 1, PG_WAIT_EXTENSION);
/* Reset the latch, loop. */
ResetLatch(MyLatch);
}
MHD_stop_daemon(daemon);
}
/*
* SQL-callable function to launch telemetry.
*/
PG_FUNCTION_INFO_V1(telemetry_server_start);
Datum
telemetry_server_start(PG_FUNCTION_ARGS)
{
unsigned short port = PG_GETARG_INT32(0);
telemetry_start_worker(port);
PG_RETURN_VOID();
}
/*
* Start telemetry worker process.
*/
static void
telemetry_start_worker(unsigned short port)
{
BackgroundWorker worker;
BackgroundWorkerHandle *handle;
BgwHandleStatus status;
pid_t pid;
MemSet(&worker, 0, sizeof(BackgroundWorker));
worker.bgw_flags = BGWORKER_SHMEM_ACCESS;
worker.bgw_start_time = BgWorkerStart_ConsistentState;
worker.bgw_main_arg = Int32GetDatum(port);
strcpy(worker.bgw_library_name, "telemetry");
strcpy(worker.bgw_function_name, "telemetry_main");
strcpy(worker.bgw_name, "telemetry server");
strcpy(worker.bgw_type, "telemetry server");
if (process_shared_preload_libraries_in_progress)
{
RegisterBackgroundWorker(&worker);
return;
}
/* must set notify PID to wait for startup */
worker.bgw_notify_pid = MyProcPid;
if (!RegisterDynamicBackgroundWorker(&worker, &handle))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
errmsg("could not register background process"),
errhint("You may need to increase max_worker_processes.")));
status = WaitForBackgroundWorkerStartup(handle, &pid);
if (status != BGWH_STARTED)
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
errmsg("could not start background process"),
errhint("More details may be available in the server log.")));
}
struct activity_stats stats;
static enum MHD_Result
telemetry_handler(void *cls,
struct MHD_Connection *connection,
const char *url,
const char *method,
const char *version,
const char *upload_data,
size_t *upload_data_size,
void **con_cls)
{
struct MHD_Response *mhd_response;
enum MHD_Result mhd_retval;
char *buf;
elog(LOG, "telemetry %s request: %s", method, url);
buf = pg_stat_dump_stats_string();
mhd_response = MHD_create_response_from_buffer(strlen(buf), (void*)buf, MHD_RESPMEM_PERSISTENT);
if (!mhd_response) {
elog(LOG, "NULL response!");
return MHD_NO;
}
mhd_retval = MHD_queue_response(connection, 200, mhd_response);
MHD_destroy_response(mhd_response);
if (mhd_retval != MHD_YES) {
elog(LOG, "mhd_no from queue_response");
return MHD_NO;
}
return MHD_YES;
}
static void telemetry_logger(void *arg, const char *fmt, va_list ap)
{
char buf[256];
int retval;
retval = vsnprintf(buf, sizeof(buf), fmt, ap);
if (retval >= sizeof(buf)) {
elog(LOG, "message buffer overflow need %d bytes", retval);
} else {
elog(LOG, "telemetry mhd log: %s", buf);
}
}