Seed a slot for multilateral involvement, even if only Herokai now

[pg_logfebe.git] / pg_logfebe.c

/*
 * pg_logfebe.c
 *
 * Implements a module to be loaded via shared_preload_libraries that,
 * should "logfebe.unix_socket" be set in postgresql.conf will cause
 * log traffic to be written to the unix socket in question on a
 * best-effort basis.
 *
 * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 * Portions Copyright (c) 2012, Heroku
 *
 */
#include "postgres.h"

#include <stdint.h>
#include <sys/un.h>
#include <unistd.h>

#include "access/xact.h"
#include "funcapi.h"
#include "lib/stringinfo.h"
#include "libpq/libpq-be.h"
#include "miscadmin.h"
#include "pg_config.h"
#include "pgtime.h"
#include "storage/proc.h"
#include "tcop/tcopprot.h"
#include "utils/elog.h"
#include "utils/guc.h"
#include "utils/ps_status.h"

/*
 * 64-bit byte-swapping, as per
 * http://stackoverflow.com/questions/809902/64-bit-ntohl-in-c
 */
#if defined(__linux__)
#include <endian.h>

#elif defined(__FreeBSD__) || defined(__NetBSD__)
#include <sys/endian.h>

#elif defined(__OpenBSD__)
#include <sys/types.h>
#define be16toh(x) betoh16(x)
#define be32toh(x) betoh32(x)
#define be64toh(x) betoh64(x)

#elif defined(__darwin__)
#include <libkern/OSByteOrder.h>
#define htobe32(x) __DARWIN_OSSwapInt32(x)
#define htobe64(x) __DARWIN_OSSwapInt64(x)
#endif

#if defined(__darwin__)
/*
 * On Macintosh, MSG_NOSIGNAL is not defined, but it has a moral
 * equivalent, SO_NOSIGPIPE.
 */
#define MSG_NOSIGNAL SO_NOSIGPIPE
#endif

PG_MODULE_MAGIC;

#define FORMATTED_TS_LEN 128

/*
 * Startup version string, e.g. "PG-9.2.4/logfebe-1", where the
 * "logfebe-1" indicates the pg_logfebe protocol version.
 */
#define PROTO_VERSION ("PG-" PG_VERSION "/logfebe-1")

/* GUC-configured destination of the log pages */
static char *logUnixSocketPath = "";
static char *ident = "";

/* Old hook storage for loading/unloading of the extension */
static emit_log_hook_type prev_emit_log_hook = NULL;

/* Used to detect if values inherited over fork need resetting. */
static int savedPid = 0;

/* Caches the formatted start time */
static char cachedBackendStartTime[FORMATTED_TS_LEN];

/* Counter for log sequence number. */
static long seqNum = 0;

/*
 * File descriptor that log records are written to.
 *
 * Is re-set if a write fails.
 */
static int outSockFd = -1;

/* Dynamic linking hooks for Postgres */
void _PG_init(void);
void _PG_fini(void);

/* Internal function definitions*/
static bool formAddr(struct sockaddr_un *dst, char *path);
static bool isLogLevelOutput(int elevel, int log_min_level);
static void appendStringInfoPtr(StringInfo dst, const char *s);
static void closeSocket(int *fd);
static void fmtLogMsg(StringInfo dst, ErrorData *edata);
static void formatLogTime(char *dst, size_t dstSz, struct timeval tv);
static void formatNow(char *dst, size_t dstSz);
static void gucOnAssignCloseInvalidate(const char *newval, void *extra);
static void logfebe_emit_log_hook(ErrorData *edata);
static void openSocket(int *dst, char *path);
static void optionalGucGet(char **dest, const char *name,
                                                   const char *shortDesc);
static void reCacheBackendStartTime(void);
static void sendOrInval(int *fd, char *payload, size_t payloadSz);


/*
 * Useful for HUP triggered reassignment: invalidate the socket, which will
 * cause path information to be evaluated when reconnection and identification
 * to be re-exchanged.
 */
static void
gucOnAssignCloseInvalidate(const char *newval, void *extra)
{
        closeSocket(&outSockFd);
}

/*
 * Procedure that wraps a bunch of boilerplate GUC options appropriate for all
 * the options used in this extension.
 */
static void
optionalGucGet(char **dest, const char *name,
                           const char *shortDesc)
{
        DefineCustomStringVariable(
                name,
                shortDesc,
                "",
                dest,
                "",
                PGC_SIGHUP,
                GUC_NOT_IN_SAMPLE,
                NULL,
                gucOnAssignCloseInvalidate,
                NULL);
}

/*
 * Form a sockaddr_un for communication, returning false if this could not be
 * completed.
 */
static bool
formAddr(struct sockaddr_un *dst, char *path)
{
        size_t len;

        dst->sun_family = AF_UNIX;
        len = strlcpy(dst->sun_path, path, sizeof dst->sun_path);

        if (len <= sizeof dst->sun_path)
        {
                /* The copy could fit, and was copied. */
                return true;
        }

        /* Truncation; dst does not contain the full passed path. */
        return false;
}

/*
 * _PG_init()                   - library load-time initialization
 *
 * DO NOT make this static nor change its name!
 *
 * Init the module, all we have to do here is getting our GUC
 */
void
_PG_init(void)
{
        /* Set up GUCs */
        optionalGucGet(&logUnixSocketPath, "logfebe.unix_socket",
                                   "Unix socket to send logs to in FEBE frames.");
        optionalGucGet(&ident, "logfebe.identity",
                                   "The identity of the installation of PostgreSQL.");

        EmitWarningsOnPlaceholders("logfebe");

        /* Install hook */
        prev_emit_log_hook = emit_log_hook;
        emit_log_hook = logfebe_emit_log_hook;
}


/*
 * Given an invalid Fd in *dst, try to open a unix socket connection to the
 * given path.
 */
static void
openSocket(int *dst, char *path)
{
        const int                       save_errno = errno;
        struct sockaddr_un      addr;
        bool                            formed;
        int                                     fd                 = -1;
        StringInfoData startup;

        /*
         * This procedure is only defined on the domain of invalidated file
         * descriptors
         */
        Assert(*dst < 0);

        /* Begin attempting connection, first by forming the address */
        formed = formAddr(&addr, path);
        if (!formed)
        {
                /* Didn't work, give up */
                goto err;
        }

        /* Get socket fd, or die */
        fd = socket(AF_LOCAL, SOCK_STREAM, 0);
        if (fd < 0)
                goto err;

        /* Connect socket to server. Or die. */
        Assert(formed);
        do
        {
                int res;

                errno = 0;
                res = connect(fd, (void *) &addr, sizeof addr);
                if (res < 0 || (errno != EINTR && errno != 0))
                        goto err;
        } while (errno == EINTR);

        /*
         * Connection established.
         *
         * Try to send start-up information as a service to the caller.  Should
         * this fail, sendOrInval will close and invalidate the socket, though.
         */
        Assert(fd >= 0);
        initStringInfo(&startup);

        /* Prepare startup: protocol version ('V') frame */
        {
                const uint32_t nVlen = htobe32((sizeof PROTO_VERSION) +
                                                                           sizeof(u_int32_t));

                appendStringInfoChar(&startup, 'V');
                appendBinaryStringInfo(&startup, (void *) &nVlen, sizeof nVlen);
                appendBinaryStringInfo(&startup, PROTO_VERSION, sizeof PROTO_VERSION);
        }

        /* Prepare startup: system identification ('I') frame */
        {
                char                    *payload;
                int                              payloadLen;
                uint32_t                 nPayloadLen;

                if (ident == NULL)
                        payload = "";
                else
                        payload = ident;

                payloadLen = strlen(payload) + sizeof '\0';
                nPayloadLen = htobe32(payloadLen + sizeof nPayloadLen);

                appendStringInfoChar(&startup, 'I');
                appendBinaryStringInfo(&startup, (void *) &nPayloadLen,
                                                           sizeof nPayloadLen);
                appendBinaryStringInfo(&startup, (void *) payload, payloadLen);
        }

        /*
         * Try to send the prepared startup packet, invaliding fd if things go
         * awry.
         */
        sendOrInval(&fd, startup.data, startup.len);
        pfree(startup.data);
        *dst = fd;
        goto exit;

err:
        /* Close and invalidate 'fd' if it got made */
        if (fd >= 0)
        {
                closeSocket(&fd);
                Assert(fd < 0);
        }

        Assert(*dst < 0);
        goto exit;

exit:
        /* Universal post-condition */
        errno = save_errno;
        return;
}

/*
 * Perform a best-effort to close and invalidate a file descriptor.
 *
 * This exists to to encapsulate EINTR handling and invalidation.
 */
static void
closeSocket(int *fd)
{
        const int save_errno = errno;

        /*
         * Close *fd and ignore EINTR, on advice from libusual's
         * "safe_close" function:
         *
         * POSIX says close() can return EINTR but fd state is "undefined"
         * later.  Seems Linux and BSDs close the fd anyway and EINTR is
         * simply informative.  Thus retry is dangerous.
         */
        close(*fd);
        *fd = -1;

        errno = save_errno;
}

static void
formatLogTime(char *dst, size_t dstSz, struct timeval tv)
{
        char                             msbuf[8];
        struct pg_tm            *tm;
        pg_time_t                        stamp_time;

        stamp_time = (pg_time_t) tv.tv_sec;
        tm = pg_localtime(&stamp_time, log_timezone);

        Assert(dstSz >= FORMATTED_TS_LEN);
        pg_strftime(dst, dstSz,
                                /* leave room for milliseconds... */
                                "%Y-%m-%d %H:%M:%S     %Z", tm);

        /* 'paste' milliseconds into place... */
        sprintf(msbuf, ".%03d", (int) (tv.tv_usec / 1000));
        strncpy(dst + 19, msbuf, 4);
}

static void
reCacheBackendStartTime(void)
{
        pg_time_t       stampTime = (pg_time_t) MyStartTime;

        /*
         * Note: we expect that guc.c will ensure that log_timezone is set up (at
         * least with a minimal GMT value) before Log_line_prefix can become
         * nonempty or CSV mode can be selected.
         */
        pg_strftime(cachedBackendStartTime, FORMATTED_TS_LEN,
                                "%Y-%m-%d %H:%M:%S %Z",
                                pg_localtime(&stampTime, log_timezone));
}

static void
formatNow(char *dst, size_t dstSz)
{
        struct timeval tv;

        gettimeofday(&tv, NULL);

        Assert(dstSz >= FORMATTED_TS_LEN);
        formatLogTime(dst, dstSz, tv);
}

/*
 * isLogLevelOutput -- is elevel logically >= log_min_level?
 *
 * We use this for tests that should consider LOG to sort out-of-order,
 * between ERROR and FATAL.  Generally this is the right thing for testing
 * whether a message should go to the postmaster log, whereas a simple >=
 * test is correct for testing whether the message should go to the client.
 */
static bool
isLogLevelOutput(int elevel, int log_min_level)
{
        if (elevel == LOG || elevel == COMMERROR)
        {
                if (log_min_level == LOG || log_min_level <= ERROR)
                        return true;
        }
        else if (log_min_level == LOG)
        {
                /* elevel != LOG */
                if (elevel >= FATAL)
                        return true;
        }
        /* Neither is LOG */
        else if (elevel >= log_min_level)
                return true;

        return false;
}

/*
 * Append a string in a special format that prepends information about
 * its NULL-ity, should it be NULL.
 */
static void
appendStringInfoPtr(StringInfo dst, const char *s)
{
        /* 'N' for NULL, 'P' for "Present" */
        if (s == NULL)
                appendStringInfoChar(dst, 'N');
        else
        {
                appendStringInfoChar(dst, 'P');
                appendStringInfoString(dst, s);
        }

        appendStringInfoChar(dst, '\0');
}

static void
fmtLogMsg(StringInfo dst, ErrorData *edata)
{
        {
                char formattedLogTime[FORMATTED_TS_LEN];

                /* timestamp with milliseconds */
                formatNow(formattedLogTime, sizeof formattedLogTime);

                /*
                 * Always present, non-nullable; don't need to write the N/P
                 * header.
                 */
                appendStringInfoString(dst, formattedLogTime);
                appendStringInfoChar(dst, '\0');
        }

        /* username */
        if (MyProcPort)
                appendStringInfoPtr(dst, MyProcPort->user_name);
        else
                appendStringInfoPtr(dst, NULL);

        /* database name */
        if (MyProcPort)
                appendStringInfoPtr(dst, MyProcPort->database_name);
        else
                appendStringInfoPtr(dst, NULL);

        /* Process id  */
        {
                uint32_t nPid = htobe32(savedPid);

                appendBinaryStringInfo(dst, (void *) &nPid, sizeof nPid);
        }

        /* Remote host and port */
        if (MyProcPort && MyProcPort->remote_host)
        {
                /* 'present' string header, since this string is nullable */
                appendStringInfoChar(dst, 'P');

                appendStringInfoString(dst, MyProcPort->remote_host);
                if (MyProcPort->remote_port && MyProcPort->remote_port[0] != '\0')
                {
                        appendStringInfoChar(dst, ':');
                        appendStringInfoString(dst, MyProcPort->remote_port);
                }

                appendStringInfoChar(dst, '\0');
        }
        else
                appendStringInfoPtr(dst, NULL);

        /* session id; non-nullable */
        appendStringInfo(dst, "%lx.%x", (long) MyStartTime, MyProcPid);
        appendStringInfoChar(dst, '\0');

        /* Line number */
        {
                uint64_t nSeqNum = htobe64(seqNum);
                appendBinaryStringInfo(dst, (void *) &nSeqNum, sizeof nSeqNum);
        }

        /* PS display */
        if (MyProcPort)
        {
                StringInfoData msgbuf;
                const char *psdisp;
                int                     displen;

                initStringInfo(&msgbuf);

                psdisp = get_ps_display(&displen);
                appendBinaryStringInfo(&msgbuf, psdisp, displen);

                appendStringInfoChar(dst, 'P');
                appendStringInfoString(dst, msgbuf.data);
                appendStringInfoChar(dst, '\0');

                pfree(msgbuf.data);
        }
        else
                appendStringInfoPtr(dst, NULL);

        /* session start timestamp */
        if (cachedBackendStartTime[0] == '\0')
        {
                /* Rebuild the cache if it was blown */
                reCacheBackendStartTime();
        }

        /* backend start time; non-nullable string */
        appendStringInfoString(dst, cachedBackendStartTime);
        appendStringInfoChar(dst, '\0');

        /*
         * Virtual transaction id
         *
         * keep VXID format in sync with lockfuncs.c
         */
        if (MyProc != NULL && MyProc->backendId != InvalidBackendId)
        {
                appendStringInfoChar(dst, 'P');
                appendStringInfo(dst, "%d/%u", MyProc->backendId, MyProc->lxid);
                appendStringInfoChar(dst, '\0');
        }
        else
                appendStringInfoPtr(dst, NULL);

        /*
         * Transaction id
         *
         * This seems to be a mistake both here and in elog.c; in particular, it's
         * not clear how the epoch would get added here.  However, leave room in
         * the protocol to fix this later by upcasting.
         */
        {
                uint64_t nTxid = htobe64((uint64) GetTopTransactionIdIfAny());

                appendBinaryStringInfo(dst, (void *) &nTxid, sizeof nTxid);
        }

        /* Error severity */
        {
                uint32_t nelevel = htobe32(edata->elevel);

                appendBinaryStringInfo(dst, (void *) &nelevel, sizeof nelevel);
        }

        /* SQL state code */
        appendStringInfoPtr(dst, unpack_sql_state(edata->sqlerrcode));

        /* errmessage */
        appendStringInfoPtr(dst, edata->message);

        /* errdetail or errdetail_log */
        if (edata->detail_log)
                appendStringInfoPtr(dst, edata->detail_log);
        else
                appendStringInfoPtr(dst, edata->detail);

        /* errhint */
        appendStringInfoPtr(dst, edata->hint);

        /* internal query */
        appendStringInfoPtr(dst, edata->internalquery);

        /* if printed internal query, print internal pos too */
        if (edata->internalpos > 0 && edata->internalquery != NULL)
        {
                uint32_t ninternalpos = htobe32(edata->internalpos);

                appendBinaryStringInfo(dst, (void *) &ninternalpos,
                                                           sizeof ninternalpos);
        }
        else
        {
                uint32_t ninternalpos = htobe32(-1);

                appendBinaryStringInfo(dst, (void *) &ninternalpos,
                                                           sizeof ninternalpos);
        }

        /* errcontext */
        appendStringInfoPtr(dst, edata->context);

        /*
         * user query --- only reported if not disabled by the caller.
         *
         * Also include query position.
         */
        if (isLogLevelOutput(edata->elevel, log_min_error_statement) &&
                debug_query_string != NULL && !edata->hide_stmt)
        {
                uint32_t nCursorPos = htobe32(edata->cursorpos);

                appendStringInfoPtr(dst, debug_query_string);
                appendBinaryStringInfo(dst, (void *) &nCursorPos, sizeof nCursorPos);
        }
        else
        {
                uint32_t nCursorPos = htobe32(-1);

                appendStringInfoPtr(dst, NULL);
                appendBinaryStringInfo(dst, (void *) &nCursorPos, sizeof nCursorPos);
        }

        /* file error location */
        if (Log_error_verbosity >= PGERROR_VERBOSE)
        {
                StringInfoData msgbuf;

                initStringInfo(&msgbuf);

                if (edata->funcname && edata->filename)
                        appendStringInfo(&msgbuf, "%s, %s:%d",
                                                         edata->funcname, edata->filename,
                                                         edata->lineno);
                else if (edata->filename)
                        appendStringInfo(&msgbuf, "%s:%d",
                                                         edata->filename, edata->lineno);

                appendStringInfoChar(dst, 'P');
                appendStringInfoString(dst, msgbuf.data);
                appendStringInfoChar(dst, '\0');

                pfree(msgbuf.data);
        }
        else
                appendStringInfoPtr(dst, NULL);

        /* application name */
        appendStringInfoPtr(dst, application_name);
}

/*
 * Send the payload or invalidate *fd.
 *
 * No confirmation of success or failure is delivered.
 */
static void
sendOrInval(int *fd, char *payload, size_t payloadSz)
{
        const int saved_errno = errno;
        ssize_t bytesWritten;

writeAgain:
        errno = 0;

        /*
         * Send, and carefully suppress SIGPIPE, which otherwise will
         * cause sendOrInval's error handling to function in since a
         * failure will come in as a signal rather than an error code.
         *
         * This is required to allow re-connection in event the server
         * closes the connection.
         */
        bytesWritten = send(*fd, payload, payloadSz, MSG_NOSIGNAL);

        /*
         * NB: Carefully perform signed-integer conversion to ssize_t;
         * otherwise the comparison delivers unintuitive results.
         */
        if (bytesWritten < (ssize_t) payloadSz)
        {
                /*
                 * Something went wrong.
                 *
                 * The ErrorData passed to the hook goes un-logged in this case (except
                 * when errno is EINTR).
                 *
                 * Because *fd is presumed a blocking socket, it is expected that
                 * whenever a full write could not be achieved that something is awry,
                 * and that the connection should abandoned.
                 */
                Assert(errno != 0);

                /* Harmless and brief; just try again */
                if (errno == EINTR)
                        goto writeAgain;

                /*
                 * Close and invalidate the socket fd; a new attempt to get a valid fd
                 * must come the next time this hook is called.
                 */
                closeSocket(fd);
        }

        errno = saved_errno;
}

static void
logfebe_emit_log_hook(ErrorData *edata)
{
        int save_errno;
        StringInfoData buf;

        /*
         * This is one of the few places where we'd rather not inherit a static
         * variable's value from the postmaster.  But since we will, reset it when
         * MyProcPid changes.
         */
        if (savedPid != MyProcPid)
        {
                savedPid = MyProcPid;

                /* Invalidate all inherited values */
                seqNum = 0;
                cachedBackendStartTime[0] = '\0';

                if (outSockFd >= 0)
                {
                        closeSocket(&outSockFd);
                }
        }

        /*
         * Increment log sequence number
         *
         * Done early on so this happens regardless if there are problems emitting
         * the log.
         */
        seqNum += 1;

        /*
         * Early exit if the socket path is not set and isn't in the format of
         * an absolute path.
         *
         * The empty identity ("ident") is a valid one, so it is not rejected in
         * the same way an empty logUnixSocketPath is.
         */
        if (logUnixSocketPath == NULL ||
                strlen(logUnixSocketPath) <= 0 || logUnixSocketPath[0] != '/')
        {
                /*
                 * Unsetting the GUCs via SIGHUP would leave a connection
                 * dangling, if it exists, close it.
                 */
                if (outSockFd >= 0)
                {
                        closeSocket(&outSockFd);
                }

                goto quickExit;
        }

        save_errno = errno;

        /*
         * Initialize StringInfoDatas early, because pfree is called
         * unconditionally at exit.
         */
        initStringInfo(&buf);

        if (outSockFd < 0)
        {
                openSocket(&outSockFd, logUnixSocketPath);

                /* Couldn't get a valid socket; give up */
                if (outSockFd < 0)
                        goto exit;
        }

        /*
         * Make room for message type byte and length header.  The length header
         * must be overwritten to the correct value at the end.
         */
        {
                const char logHdr[5] = {'L', '\0', '\0', '\0', '\0'};

                appendBinaryStringInfo(&buf, logHdr, sizeof logHdr);
        }

        /*
         * Format the output, and figure out how long it is, and frame it
         * for the protocol.
         */
        {
                uint32_t *msgSize;

                fmtLogMsg(&buf, edata);

                /*
                 * Check that buf is prepared properly, with enough space and
                 * the placeholder length expected.
                 */
                Assert(buf.len > 5);
                Assert(buf.data[0] == 'L');

                msgSize = (uint32_t *)(buf.data + 1);
                Assert(*msgSize == 0);

                /*
                 * Fill in *msgSize: buf contains the msg header, which is not
                 * included in length; subract and byte-swap to paste the
                 * right length into place.
                 */
                *msgSize = htobe32(buf.len - 1);
        }

        /* Finally: try to send the constructed message */
        sendOrInval(&outSockFd, buf.data, buf.len);

exit:
        pfree(buf.data);
        errno = save_errno;

quickExit:
        /* Call a previous hook, should it exist */
        if (prev_emit_log_hook != NULL)
                prev_emit_log_hook(edata);
}

/*
 * Module unload callback
 */
void
_PG_fini(void)
{
        /* Uninstall hook */
        emit_log_hook = prev_emit_log_hook;
}