#include "sys_defs.h"
#ifdef HAS_PGSQL
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <syslog.h>
#include <time.h>
#include <postgres_ext.h>
#include <libpq-fe.h>
#include "dict.h"
#include "msg.h"
#include "mymalloc.h"
#include "argv.h"
#include "vstring.h"
#include "split_at.h"
#include "find_inet.h"
#include "myrand.h"
#include "events.h"
#include "stringops.h"
#include "cfg_parser.h"
#include "db_common.h"
#include "dict_pgsql.h"
#define STATACTIVE (1<<0)
#define STATFAIL (1<<1)
#define STATUNTRIED (1<<2)
#define TYPEUNIX (1<<0)
#define TYPEINET (1<<1)
#define RETRY_CONN_MAX 100
#define RETRY_CONN_INTV 60
#define IDLE_CONN_INTV 60
typedef struct {
PGconn *db;
char *hostname;
char *name;
char *port;
unsigned type;
unsigned stat;
time_t ts;
} HOST;
typedef struct {
int len_hosts;
HOST **db_hosts;
} PLPGSQL;
typedef struct {
DICT dict;
CFG_PARSER *parser;
char *query;
char *result_format;
void *ctx;
int expansion_limit;
char *username;
char *password;
char *dbname;
char *table;
ARGV *hosts;
PLPGSQL *pldb;
HOST *active_host;
} DICT_PGSQL;
#define PGSQL_RES PGresult
static PLPGSQL *plpgsql_init(ARGV *);
static PGSQL_RES *plpgsql_query(DICT_PGSQL *, const char *, VSTRING *, char *,
char *, char *);
static void plpgsql_dealloc(PLPGSQL *);
static void plpgsql_close_host(HOST *);
static void plpgsql_down_host(HOST *);
static void plpgsql_connect_single(HOST *, char *, char *, char *);
static const char *dict_pgsql_lookup(DICT *, const char *);
DICT *dict_pgsql_open(const char *, int, int);
static void dict_pgsql_close(DICT *);
static HOST *host_init(const char *);
static void dict_pgsql_quote(DICT *dict, const char *name, VSTRING *result)
{
DICT_PGSQL *dict_pgsql = (DICT_PGSQL *) dict;
HOST *active_host = dict_pgsql->active_host;
char *myname = "dict_pgsql_quote";
size_t len = strlen(name);
size_t buflen;
int err = 1;
if (active_host == 0)
msg_panic("%s: bogus dict_pgsql->active_host", myname);
if (len > (SSIZE_T_MAX - VSTRING_LEN(result) - 1) / 2)
msg_panic("%s: arithmetic overflow in %lu+2*%lu+1",
myname, (unsigned long) VSTRING_LEN(result),
(unsigned long) len);
buflen = 2 * len + 1;
if (active_host->stat == STATFAIL)
return;
VSTRING_SPACE(result, buflen);
PQescapeStringConn(active_host->db, vstring_end(result), name, len, &err);
if (err == 0) {
VSTRING_SKIP(result);
} else {
msg_warn("dict pgsql: (host %s) cannot escape input string: %s",
active_host->hostname, PQerrorMessage(active_host->db));
active_host->stat = STATFAIL;
VSTRING_TERMINATE(result);
}
}
static const char *dict_pgsql_lookup(DICT *dict, const char *name)
{
const char *myname = "dict_pgsql_lookup";
PGSQL_RES *query_res;
DICT_PGSQL *dict_pgsql;
static VSTRING *query;
static VSTRING *result;
int i;
int j;
int numrows;
int numcols;
int expansion;
const char *r;
int domain_rc;
dict_pgsql = (DICT_PGSQL *) dict;
#define INIT_VSTR(buf, len) do { \
if (buf == 0) \
buf = vstring_alloc(len); \
VSTRING_RESET(buf); \
VSTRING_TERMINATE(buf); \
} while (0)
INIT_VSTR(query, 10);
INIT_VSTR(result, 10);
dict->error = 0;
if (dict->flags & DICT_FLAG_FOLD_FIX) {
if (dict->fold_buf == 0)
dict->fold_buf = vstring_alloc(10);
vstring_strcpy(dict->fold_buf, name);
name = lowercase(vstring_str(dict->fold_buf));
}
if ((domain_rc = db_common_check_domain(dict_pgsql->ctx, name)) == 0) {
if (msg_verbose)
msg_info("%s: Skipping lookup of '%s'", myname, name);
return (0);
}
if (domain_rc < 0)
DICT_ERR_VAL_RETURN(dict, domain_rc, (char *) 0);
if (!db_common_expand(dict_pgsql->ctx, dict_pgsql->query,
name, 0, query, 0))
return (0);
if ((query_res = plpgsql_query(dict_pgsql, name, query,
dict_pgsql->dbname,
dict_pgsql->username,
dict_pgsql->password)) == 0) {
dict->error = DICT_ERR_RETRY;
return 0;
}
numrows = PQntuples(query_res);
if (msg_verbose)
msg_info("%s: retrieved %d rows", myname, numrows);
if (numrows == 0) {
PQclear(query_res);
return 0;
}
numcols = PQnfields(query_res);
for (expansion = i = 0; i < numrows && dict->error == 0; i++) {
for (j = 0; j < numcols; j++) {
r = PQgetvalue(query_res, i, j);
if (db_common_expand(dict_pgsql->ctx, dict_pgsql->result_format,
r, name, result, 0)
&& dict_pgsql->expansion_limit > 0
&& ++expansion > dict_pgsql->expansion_limit) {
msg_warn("%s: %s: Expansion limit exceeded for key: '%s'",
myname, dict_pgsql->parser->name, name);
dict->error = DICT_ERR_RETRY;
break;
}
}
}
PQclear(query_res);
r = vstring_str(result);
return ((dict->error == 0 && *r) ? r : 0);
}
static int dict_pgsql_check_stat(HOST *host, unsigned stat, unsigned type,
time_t t)
{
if ((host->stat & stat) && (!type || host->type & type)) {
if (host->stat == STATFAIL && host->ts > 0 && host->ts >= t)
return 0;
return 1;
}
return 0;
}
static HOST *dict_pgsql_find_host(PLPGSQL *PLDB, unsigned stat, unsigned type)
{
time_t t;
int count = 0;
int idx;
int i;
t = time((time_t *) 0);
for (i = 0; i < PLDB->len_hosts; i++) {
if (dict_pgsql_check_stat(PLDB->db_hosts[i], stat, type, t))
count++;
}
if (count) {
idx = (count > 1) ?
1 + count * (double) myrand() / (1.0 + RAND_MAX) : 1;
for (i = 0; i < PLDB->len_hosts; i++) {
if (dict_pgsql_check_stat(PLDB->db_hosts[i], stat, type, t) &&
--idx == 0)
return PLDB->db_hosts[i];
}
}
return 0;
}
static HOST *dict_pgsql_get_active(PLPGSQL *PLDB, char *dbname,
char *username, char *password)
{
const char *myname = "dict_pgsql_get_active";
HOST *host;
int count = RETRY_CONN_MAX;
if ((host = dict_pgsql_find_host(PLDB, STATACTIVE, TYPEUNIX)) != NULL ||
(host = dict_pgsql_find_host(PLDB, STATACTIVE, TYPEINET)) != NULL) {
if (msg_verbose)
msg_info("%s: found active connection to host %s", myname,
host->hostname);
return host;
}
while (--count > 0 &&
((host = dict_pgsql_find_host(PLDB, STATUNTRIED | STATFAIL,
TYPEUNIX)) != NULL ||
(host = dict_pgsql_find_host(PLDB, STATUNTRIED | STATFAIL,
TYPEINET)) != NULL)) {
if (msg_verbose)
msg_info("%s: attempting to connect to host %s", myname,
host->hostname);
plpgsql_connect_single(host, dbname, username, password);
if (host->stat == STATACTIVE)
return host;
}
return 0;
}
static void dict_pgsql_event(int unused_event, void *context)
{
HOST *host = (HOST *) context;
if (host->db)
plpgsql_close_host(host);
}
static PGSQL_RES *plpgsql_query(DICT_PGSQL *dict_pgsql,
const char *name,
VSTRING *query,
char *dbname,
char *username,
char *password)
{
PLPGSQL *PLDB = dict_pgsql->pldb;
HOST *host;
PGSQL_RES *res = 0;
ExecStatusType status;
while ((host = dict_pgsql_get_active(PLDB, dbname, username, password)) != NULL) {
dict_pgsql->active_host = host;
VSTRING_RESET(query);
VSTRING_TERMINATE(query);
db_common_expand(dict_pgsql->ctx, dict_pgsql->query,
name, 0, query, dict_pgsql_quote);
dict_pgsql->active_host = 0;
if (host->stat == STATFAIL) {
plpgsql_down_host(host);
continue;
}
if ((res = PQexec(host->db, vstring_str(query))) != 0) {
switch ((status = PQresultStatus(res))) {
case PGRES_TUPLES_OK:
case PGRES_COMMAND_OK:
if (msg_verbose)
msg_info("dict_pgsql: successful query from host %s",
host->hostname);
event_request_timer(dict_pgsql_event, (void *) host,
IDLE_CONN_INTV);
return (res);
case PGRES_FATAL_ERROR:
msg_warn("pgsql query failed: fatal error from host %s: %s",
host->hostname, PQresultErrorMessage(res));
break;
case PGRES_BAD_RESPONSE:
msg_warn("pgsql query failed: protocol error, host %s",
host->hostname);
break;
default:
msg_warn("pgsql query failed: unknown code 0x%lx from host %s",
(unsigned long) status, host->hostname);
break;
}
} else {
msg_warn("pgsql query failed: fatal error from host %s: %s",
host->hostname, PQerrorMessage(host->db));
}
if (res != 0)
PQclear(res);
plpgsql_down_host(host);
}
return (0);
}
static void plpgsql_connect_single(HOST *host, char *dbname, char *username, char *password)
{
if ((host->db = PQsetdbLogin(host->name, host->port, NULL, NULL,
dbname, username, password)) == NULL
|| PQstatus(host->db) != CONNECTION_OK) {
msg_warn("connect to pgsql server %s: %s",
host->hostname, PQerrorMessage(host->db));
plpgsql_down_host(host);
return;
}
if (msg_verbose)
msg_info("dict_pgsql: successful connection to host %s",
host->hostname);
if (PQsetClientEncoding(host->db, "LATIN1") != 0) {
msg_warn("dict_pgsql: cannot set the encoding to LATIN1, skipping %s",
host->hostname);
plpgsql_down_host(host);
return;
}
host->stat = STATACTIVE;
}
static void plpgsql_close_host(HOST *host)
{
if (host->db)
PQfinish(host->db);
host->db = 0;
host->stat = STATUNTRIED;
}
static void plpgsql_down_host(HOST *host)
{
if (host->db)
PQfinish(host->db);
host->db = 0;
host->ts = time((time_t *) 0) + RETRY_CONN_INTV;
host->stat = STATFAIL;
event_cancel_timer(dict_pgsql_event, (void *) host);
}
static void pgsql_parse_config(DICT_PGSQL *dict_pgsql, const char *pgsqlcf)
{
const char *myname = "pgsql_parse_config";
CFG_PARSER *p = dict_pgsql->parser;
char *hosts;
VSTRING *query;
char *select_function;
dict_pgsql->username = cfg_get_str(p, "user", "", 0, 0);
dict_pgsql->password = cfg_get_str(p, "password", "", 0, 0);
dict_pgsql->dbname = cfg_get_str(p, "dbname", "", 1, 0);
dict_pgsql->result_format = cfg_get_str(p, "result_format", "%s", 1, 0);
dict_pgsql->expansion_limit = cfg_get_int(dict_pgsql->parser,
"expansion_limit", 0, 0, 0);
if ((dict_pgsql->query = cfg_get_str(p, "query", 0, 0, 0)) == 0) {
query = vstring_alloc(64);
select_function = cfg_get_str(p, "select_function", 0, 0, 0);
if (select_function != 0) {
vstring_sprintf(query, "SELECT %s('%%s')", select_function);
myfree(select_function);
} else
db_common_sql_build_query(query, p);
dict_pgsql->query = vstring_export(query);
}
dict_pgsql->ctx = 0;
(void) db_common_parse(&dict_pgsql->dict, &dict_pgsql->ctx,
dict_pgsql->query, 1);
(void) db_common_parse(0, &dict_pgsql->ctx, dict_pgsql->result_format, 0);
db_common_parse_domain(p, dict_pgsql->ctx);
if (db_common_dict_partial(dict_pgsql->ctx))
dict_pgsql->dict.flags |= DICT_FLAG_PATTERN;
else
dict_pgsql->dict.flags |= DICT_FLAG_FIXED;
if (dict_pgsql->dict.flags & DICT_FLAG_FOLD_FIX)
dict_pgsql->dict.fold_buf = vstring_alloc(10);
hosts = cfg_get_str(p, "hosts", "", 0, 0);
dict_pgsql->hosts = argv_split(hosts, CHARS_COMMA_SP);
if (dict_pgsql->hosts->argc == 0) {
argv_add(dict_pgsql->hosts, "localhost", ARGV_END);
argv_terminate(dict_pgsql->hosts);
if (msg_verbose)
msg_info("%s: %s: no hostnames specified, defaulting to '%s'",
myname, pgsqlcf, dict_pgsql->hosts->argv[0]);
}
myfree(hosts);
}
DICT *dict_pgsql_open(const char *name, int open_flags, int dict_flags)
{
DICT_PGSQL *dict_pgsql;
CFG_PARSER *parser;
if (open_flags != O_RDONLY)
return (dict_surrogate(DICT_TYPE_PGSQL, name, open_flags, dict_flags,
"%s:%s map requires O_RDONLY access mode",
DICT_TYPE_PGSQL, name));
if ((parser = cfg_parser_alloc(name)) == 0)
return (dict_surrogate(DICT_TYPE_PGSQL, name, open_flags, dict_flags,
"open %s: %m", name));
dict_pgsql = (DICT_PGSQL *) dict_alloc(DICT_TYPE_PGSQL, name,
sizeof(DICT_PGSQL));
dict_pgsql->dict.lookup = dict_pgsql_lookup;
dict_pgsql->dict.close = dict_pgsql_close;
dict_pgsql->dict.flags = dict_flags;
dict_pgsql->parser = parser;
pgsql_parse_config(dict_pgsql, name);
dict_pgsql->active_host = 0;
dict_pgsql->pldb = plpgsql_init(dict_pgsql->hosts);
if (dict_pgsql->pldb == NULL)
msg_fatal("couldn't initialize pldb!\n");
dict_pgsql->dict.owner = cfg_get_owner(dict_pgsql->parser);
return (DICT_DEBUG (&dict_pgsql->dict));
}
static PLPGSQL *plpgsql_init(ARGV *hosts)
{
PLPGSQL *PLDB;
int i;
PLDB = (PLPGSQL *) mymalloc(sizeof(PLPGSQL));
PLDB->len_hosts = hosts->argc;
PLDB->db_hosts = (HOST **) mymalloc(sizeof(HOST *) * hosts->argc);
for (i = 0; i < hosts->argc; i++)
PLDB->db_hosts[i] = host_init(hosts->argv[i]);
return PLDB;
}
static HOST *host_init(const char *hostname)
{
const char *myname = "pgsql host_init";
HOST *host = (HOST *) mymalloc(sizeof(HOST));
const char *d = hostname;
host->db = 0;
host->hostname = mystrdup(hostname);
host->stat = STATUNTRIED;
host->ts = 0;
if (strncmp(d, "unix:", 5) == 0 || strncmp(d, "inet:", 5) == 0)
d += 5;
host->name = mystrdup(d);
host->port = split_at_right(host->name, ':');
if (host->name[0] && host->name[0] != '/')
host->type = TYPEINET;
else
host->type = TYPEUNIX;
if (msg_verbose > 1)
msg_info("%s: host=%s, port=%s, type=%s", myname, host->name,
host->port ? host->port : "",
host->type == TYPEUNIX ? "unix" : "inet");
return host;
}
static void dict_pgsql_close(DICT *dict)
{
DICT_PGSQL *dict_pgsql = (DICT_PGSQL *) dict;
plpgsql_dealloc(dict_pgsql->pldb);
cfg_parser_free(dict_pgsql->parser);
myfree(dict_pgsql->username);
myfree(dict_pgsql->password);
myfree(dict_pgsql->dbname);
myfree(dict_pgsql->query);
myfree(dict_pgsql->result_format);
if (dict_pgsql->hosts)
argv_free(dict_pgsql->hosts);
if (dict_pgsql->ctx)
db_common_free_ctx(dict_pgsql->ctx);
if (dict->fold_buf)
vstring_free(dict->fold_buf);
dict_free(dict);
}
static void plpgsql_dealloc(PLPGSQL *PLDB)
{
int i;
for (i = 0; i < PLDB->len_hosts; i++) {
event_cancel_timer(dict_pgsql_event, (void *) (PLDB->db_hosts[i]));
if (PLDB->db_hosts[i]->db)
PQfinish(PLDB->db_hosts[i]->db);
myfree(PLDB->db_hosts[i]->hostname);
myfree(PLDB->db_hosts[i]->name);
myfree((void *) PLDB->db_hosts[i]);
}
myfree((void *) PLDB->db_hosts);
myfree((void *) (PLDB));
}
#endif