Merge pull request #15 from mkindahl/copyable-worker-args

Pass worker data correctly

expected/inval.out

@@ -3,7 +3,6 @@ CREATE TABLE db_worker.cpu(_time timestamptz, _tags jsonb, _fields jsonb);
 \set VERBOSITY terse
 \x on
 SELECT pg_sleep(1) FROM worker_launch('db_worker', 4711::text);
-NOTICE:  background worker started
 -[ RECORD 1 ]
 pg_sleep | 
 

expected/worker.out

@@ -5,7 +5,6 @@ CREATE TABLE db_worker.system(_time timestamp, host text, uptime int, _tags json
 \set VERBOSITY terse
 \x on
 SELECT pg_sleep(1) FROM worker_launch('db_worker', 4711::text);
-NOTICE:  background worker started
 -[ RECORD 1 ]
 pg_sleep | 
 

influx.c

@@ -150,12 +150,19 @@ static void StartBackgroundWorkers(const char *database_name,
                                    const char *role_name,
                                    const char *service_name, int worker_count) {
   MemoryContext oldcontext = MemoryContextSwitchTo(TopMemoryContext);
-  WorkerArgs args = {.namespace = schema_name ? pstrdup(schema_name) : NULL,
-                     .database = database_name ? pstrdup(database_name) : NULL,
-                     .role = role_name ? pstrdup(role_name) : NULL,
-                     .service = service_name ? pstrdup(service_name) : NULL};
+  WorkerArgs args = {0};
   int i;
   BackgroundWorker worker;
+
+  if (schema_name)
+    strncpy(args.namespace, schema_name, sizeof(args.namespace));
+  if (database_name)
+    strncpy(args.database, database_name, sizeof(args.database));
+  if (role_name)
+    strncpy(args.role, role_name, sizeof(args.role));
+  if (service_name)
+    strncpy(args.service, service_name, sizeof(args.service));
+
   elog(LOG, "starting influx workers");
 
   InfluxWorkerInit(&worker, &args);

metric.c

@@ -119,8 +119,6 @@ static void PrepareRecord(Relation rel, Oid *argtypes, PreparedInsert record) {
   const Oid relid = RelationGetRelid(rel);
   int i;
 
-  elog(NOTICE, "preparing statement for %s", SPI_getrelname(rel));
-
   /* Using the tuple descriptor and the parsed package, build the
    * insert statement and collect the null array for the prepare
    * call. */

network.c

@@ -80,9 +80,9 @@ struct SocketMethod UdpSendSocket = {
 int SocketPort(struct sockaddr* addr, socklen_t addrlen) {
   switch (addr->sa_family) {
     case AF_INET:
-      return ((struct sockaddr_in*)addr)->sin_port;
+      return ntohs(((struct sockaddr_in*)addr)->sin_port);
     case AF_INET6:
-      return ((struct sockaddr_in6*)addr)->sin6_port;
+      return ntohs(((struct sockaddr_in6*)addr)->sin6_port);
     default:
       return -1;
   }

worker.c

@@ -161,12 +161,13 @@ void InfluxWorkerMain(Datum arg) {
      crash. */
   namespace_id = get_namespace_oid(args->namespace, false);
 
-  ereport(LOG,
-          (errmsg("worker listening on port %d (service %s)",
-                  SocketPort((struct sockaddr *)&sockaddr, sizeof(sockaddr)),
-                  args->service),
-           errdetail("database=%s, namespace=%s, user=%s", args->database,
-                     args->namespace, args->role)));
+  ereport(
+      LOG,
+      (errmsg("worker listening on port %d",
+              SocketPort((struct sockaddr *)&sockaddr, sizeof(sockaddr))),
+       errdetail(
+           "Connected to database %s as user %s. Metrics written to schema %s.",
+           args->database, args->role, args->namespace)));
 
   pgstat_report_activity(STATE_RUNNING, "reading events");
 
@@ -255,15 +256,18 @@ Datum worker_launch(PG_FUNCTION_ARGS) {
   BackgroundWorkerHandle *handle;
   BgwHandleStatus status;
   pid_t pid;
-  WorkerArgs args = {.service = service};
+  WorkerArgs args = {0};
 
   /* Check that we have a valid namespace id */
   if (get_namespace_name(nspid) == NULL)
     ereport(ERROR, (errcode(ERRCODE_UNDEFINED_SCHEMA),
                     errmsg("schema with OID %d does not exist", nspid)));
 
-  args.namespace = get_namespace_name(nspid);
-  args.database = get_database_name(MyDatabaseId);
+  strncpy(args.role, GetUserNameFromId(GetUserId(), true), sizeof(args.role));
+  strncpy(args.service, service, sizeof(args.service));
+  strncpy(args.namespace, get_namespace_name(nspid), sizeof(args.namespace));
+  strncpy(args.database, get_database_name(MyDatabaseId),
+          sizeof(args.database));
 
   InfluxWorkerInit(&worker, &args);
 
@@ -293,8 +297,7 @@ Datum worker_launch(PG_FUNCTION_ARGS) {
              errhint("Kill all remaining database processes and restart the "
                      "database.")));
 
-  ereport(NOTICE,
-          (errmsg("background worker started"), errdetail("pid=%d", pid)));
+  ereport(LOG, (errmsg("background worker started"), errdetail("pid=%d", pid)));
 
   Assert(status == BGWH_STARTED);
   PG_RETURN_INT32(pid);

worker.h

@@ -25,10 +25,10 @@
 #define INFLUX_FUNCTION_NAME "InfluxWorkerMain"
 
 typedef struct WorkerArgs {
-  const char *role;
-  const char *namespace;
-  const char *database;
-  const char *service;
+  char role[32];
+  char namespace[32];
+  char database[32];
+  char service[32];
 } WorkerArgs;
 
 void InfluxWorkerInit(BackgroundWorker *worker, WorkerArgs *args);