Changeset 5360a1ee7f4ade3bd6299e566204ee1820aeffa3

Timestamp:

10/19/09 01:22:39 (4 years ago)

Author:

Theo Schlossnagle <jesus@omniti.com>

git-committer:

Theo Schlossnagle <jesus@omniti.com> 1255915359 +0000

git-parent:

[08e498f88b47f53edf390c85d08577fa1d7686ab]

git-author:

Theo Schlossnagle <jesus@omniti.com> 1255915359 +0000

Message:

hefty patch. First part of stratcond support for horizontal partitioning of storage. refs #150

Files:

• src/stratcon.conf.in (modified) (4 diffs)
• src/stratcon_datastore.c (modified) (24 diffs)
• src/stratcon_datastore.h (modified) (1 diff)
• src/stratcon_iep.c (modified) (7 diffs)
• src/stratcon_iep.h (modified) (1 diff)
• src/stratcon_jlog_streamer.c (modified) (2 diffs)
• src/stratcon_jlog_streamer.h (modified) (1 diff)
• src/stratcon_realtime_http.c (modified) (2 diffs)
• src/stratcond.c (modified) (1 diff)

src/stratcon.conf.in

@@ -9 +9 @@
-      <log name="error/iep"/>
-      <log name="error/eventer" disabled="true"/>
+      <log name="error/datastore" disabled="true"/>
-      <log name="debug/eventer" disabled="true"/>
-    </console_output>
@@ -31 +32 @@
-  </noits>
-
-fals
+tru
-    <start directory="%iepdbdir%"
-           command="%iepbindir%/run-iep.sh" />
@@ -62 +63 @@
-
-  <database>
+    <journal>
+      <path>/var/log/stratcon.persist</path>
+    </journal>
-    <dbconfig>
-      <host>localhost</host>
@@ -69 +73 @@
-    </dbconfig>
-    <statements>
+      <!-- These are optional and used for stuff like setting search paths -->
+      <!--
+      <metanodepostconnect><![CDATA[
+        SELECT do_some_magic();
+      ]]></metanodepostconnect>
+      <storagepostconnect><![CDATA[
+        SELECT do_some_magic($1,$2);
+      ]]></storagepostconnect>
+      -->
-      <allchecks><![CDATA[
-        SELECT remote_address, id, target, module, name
-stratcon.mv_loading_dock_check_s
+noit.get_checks()
-      ]]></allchecks>
-      <findcheck><![CDATA[
-
-
-
+
+
-      ]]></findcheck>
+      <allstoragenodes><![CDATA[
+        SELECT storage_node_id, fqdn, dsn
+          FROM stratcon.storage_node
+      ]]></allstoragenodes>
+      <findstoragenode><![CDATA[
+        SELECT fqdn, dsn
+          FROM stratcon.storage_node
+         WHERE storage_node_id = $1
+      ]]></findstoragenode>
+      <mapallchecks><![CDATA[
+        SELECT id, sid, storage_node_id, fqdn, dsn
+          FROM stratcon.map_uuid_to_sid LEFT JOIN stratcon.storage_node USING (storage_node_id)
+      ]]></mapallchecks>
+      <mapchecktostoragenode><![CDATA[
+        SELECT o_storage_node_id as storage_node_id, o_sid as sid, o_fqdn as fqdn, o_dsn as dsn from stratcon.map_uuid_to_sid($1,$2)
+      ]]></mapchecktostoragenode>
-      <check><![CDATA[
-        INSERT INTO stratcon.loading_dock_check_s

src/stratcon_datastore.c

@@ -1 +1 @@
-/*
-
+-2009
- * All rights reserved.
- *
@@ -35 +35 @@
-#include "utils/noit_log.h"
-#include "utils/noit_b64.h"
+#include "utils/noit_str.h"
+#include "utils/noit_mkdir.h"
-#include "stratcon_datastore.h"
-#include "stratcon_realtime_http.h"
@@ -41 +43 @@
-#include "noit_check.h"
-#include <unistd.h>
+#include <fcntl.h>
-#include <netinet/in.h>
-#include <sys/un.h>
+#include <dirent.h>
-#include <arpa/inet.h>
+#include <sys/mman.h>
-#include <libpq-fe.h>
-#include <zlib.h>
-#include <assert.h>
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-static struct datastore_onlooker_list {
-
+
+
-  struct datastore_onlooker_list *next;
-} *onlookers = NULL;
@@ -73 +86 @@
-      goto bad_row; \
-} while(0)
+
+struct conn_q;
+
+typedef struct {
+  char            *queue_name; /* the key fqdn+remote_sn */
+  eventer_jobq_t  *jobq;
+  struct conn_q   *head;
+  pthread_mutex_t  lock;
+  pthread_cond_t   cv;
+  int              ttl;
+  int              in_pool;
+  int              outstanding;
+  int              max_allocated;
+  int              max_in_pool;
+} conn_pool;
+typedef struct conn_q {
+  time_t           last_use;
+  char            *dsn;        /* Pg connect string */
+  char            *remote_str; /* the IP of the noit*/
+  char            *remote_cn;  /* the Cert CN of the noit */
+  char            *fqdn;       /* the fqdn of the storage node */
+  conn_pool       *pool;
+  struct conn_q   *next;
+  /* Postgres specific stuff */
+  PGconn          *dbh;
+} conn_q;
+
-
-#define MAX_PARAMS 8
@@ -89 +129 @@
-  POSTGRES_PARTS
-} ds_single_detail;
-ds_job_detail 
+
-  /* Postgres specific stuff */
-  POSTGRES_PARTS
-
-  char *data;  /* The raw string, NULL means the stream is done -- commit. */
-  struct realtime_tracker *rt;
-
+
+
+
+
+
+
+
-  int problematic;
-
-
-
+
+
-
-typedef struct {
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-static int stratcon_database_connect(conn_q *cq);
@@ -120 +169 @@
-}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
-noit_hash_table ds_conns;
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-  if(remote) {
+    int len = 0;
-    switch(remote->sa_family) {
-      case AF_INET:
-        len = sizeof(struct sockaddr_in);
-        inet_ntop(remote->sa_family, &((struct sockaddr_in *)remote)->sin_addr,
-queue_name + strlen("datastore_")
+name
-        break;
-      case AF_INET6:
-       len = sizeof(struct sockaddr_in6);
-        inet_ntop(remote->sa_family, &((struct sockaddr_in6 *)remote)->sin6_addr,
-queue_name + strlen("datastore_")
+name
-       break;
-      case AF_UNIX:
-        len = SUN_LEN(((struct sockaddr_un *)remote));
-queue_name, sizeof(queue_name), "datastore_
+name, sizeof(name), "
-        break;
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-  }
-  else {
-
-
-
-
-
-
-
+
+
+
+
-  cq = calloc(1, sizeof(*cq));
-
-
-
-
-
-
-
-
+
+
+
+
+
-  return cq;
+}
+static conn_q *
+get_conn_q_for_metanode() {
+  return get_conn_q_for_remote(NULL,NULL,NULL,NULL);
-}
-
@@ -243 +439 @@
-  if(d->rv != PGRES_COMMAND_OK && \
-     d->rv != PGRES_TUPLES_OK) { \
-noit_erro
+ds_er
-          d->rv, PQresultErrorMessage(d->res), cmd); \
-    PQclear(d->res); \
@@ -262 +458 @@
-  if(d->rv != PGRES_COMMAND_OK && \
-     d->rv != PGRES_TUPLES_OK) { \
-noit_erro
+ds_er
-          d->rv, PQresultErrorMessage(d->res), cmdbuf, \
-          (long long unsigned)whence); \
@@ -274 +470 @@
-                                    struct timeval *now) {
-  conn_q *cq = closure;
-job
+single
-  int i, row_count = 0, good = 0;
-  char buff[1024];
@@ -315 +511 @@
-
-    /* stratcon_iep_line_processor takes an allocated operand and frees it */
-strdup(buff)
+NULL, strdup(buff), NULL
-    good++;
-  }
-  noitL(noit_error, "Staged %d/%d remembered checks into IEP\n", good, row_count);
- bad_row:
-PQclear(d->res
+free_params((ds_single_detail *)d
-  free(d);
+  release_conn_q(cq);
-  return 0;
-}
@@ -328 +525 @@
-  eventer_t e;
-  conn_q *cq;
-__get_conn_q_for_remote(NULL
+get_conn_q_for_metanode(
-
-  e = eventer_alloc();
@@ -334 +531 @@
-  e->callback = stratcon_datastore_asynch_drive_iep;
-  e->closure = cq;
-
+pool->
-}
-execute_outcome_t
-job
+rt
-  char *val;
-  int row_count;
-
-
+
+
-  GET_QUERY(check_find);
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-  return DS_EXEC_SUCCESS;
- bad_row:
-  return DS_EXEC_ROW_FAILED;
-}
-execute_outcome_t
-struct sockaddr *r, ds_job
+const char *r, ds_line
-  int type, len;
-  char *final_buff;
-;
+
-  char *token;
+  char raddr_blank[1] = "";
+  const char *raddr;
-
-  type = d->data[0];
+  raddr = r ? r : raddr_blank;
-
-  /* Parse the log line, but only if we haven't already */
-  if(!d->nparams) {
-    char raddr[128];
-    char *scp, *ecp;
-
-    /* setup our remote address */
-    raddr[0] = '\0';
-    switch(r->sa_family) {
-      case AF_INET:
-        inet_ntop(AF_INET, &(((struct sockaddr_in *)r)->sin_addr),
-                  raddr, sizeof(raddr));
-        break;
-      case AF_INET6:
-        inet_ntop(AF_INET6, &(((struct sockaddr_in6 *)r)->sin6_addr),
-                  raddr, sizeof(raddr));
-        break;
-      default:
-        noitL(noit_error, "remote address of family %d\n", r->sa_family);
-    }
- 
-    scp = d->data;
-#define PROCESS_NEXT_FIELD(t,l) do { \
@@ -547 +746 @@
-}
-static int
+stratcon_database_post_connect(conn_q *cq) {
+  int rv = 0;
+  ds_single_detail _d = { 0 }, *d = &_d;
+  if(cq->remote_str) {
+    char *remote_str, *remote_cn;
+    /* This is the silly way we get null's in through our declare_param_str */
+    remote_str = cq->remote_str ? cq->remote_str : "[[null]]";
+    remote_cn = cq->remote_cn ? cq->remote_cn : "[[null]]";
+    /* This is a storage node, it gets the storage node post_connect */
+    GET_QUERY(storage_post_connect);
+    rv = -1; /* now we're serious */
+    DECLARE_PARAM_STR(remote_str, strlen(remote_str));
+    DECLARE_PARAM_STR(remote_cn, strlen(remote_cn));
+    PG_EXEC(storage_post_connect);
+    PQclear(d->res);
+    rv = 0;
+  }
+  else {
+    /* Metanode post_connect */
+    GET_QUERY(metanode_post_connect);
+    rv = -1; /* now we're serious */
+    PG_EXEC(metanode_post_connect);
+    PQclear(d->res);
+    rv = 0;
+  }
+ bad_row:
+  free_params(d);
+  if(rv == -1) {
+    /* Post-connect intentions are serious and fatal */
+    PQfinish(cq->dbh);
+    cq->dbh = NULL;
+  }
+  return rv;
+}
+static int
-stratcon_database_connect(conn_q *cq) {
-dsn
+*dsn, dsn_meta
-  noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
-  const char *k, *v;
@@ -554 +788 @@
-  noit_hash_table *t;
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-  if(cq->dbh) {
-    if(PQstatus(cq->dbh) == CONNECTION_OK) return 0;
-    PQreset(cq->dbh);
+    if(stratcon_database_post_connect(cq)) return -1;
-    if(PQstatus(cq->dbh) == CONNECTION_OK) return 0;
-    noitL(noit_error, "Error reconnecting to database: '%s'\nError: %s\n",
@@ -576 +815 @@
-  cq->dbh = PQconnectdb(dsn);
-  if(!cq->dbh) return -1;
+  if(stratcon_database_post_connect(cq)) return -1;
-  if(PQstatus(cq->dbh) == CONNECTION_OK) return 0;
-  noitL(noit_error, "Error connection to database: '%s'\nError: %s\n",
@@ -625 +865 @@
-stratcon_datastore_asynch_lookup(eventer_t e, int mask, void *closure,
-                                 struct timeval *now) {
-conn_q *cq
-ds_job_detail *current, *next
+ds_rt_detail *dsjd
+conn_q *cq = dsjd->cq
-  if(!(mask & EVENTER_ASYNCH_WORK)) return 0;
-
-  if(!cq->head) return 0; 
-
-  stratcon_database_connect(cq);
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
-  return 0;
+}
+static const char *
+get_dsn_from_storagenode_id(int id, int can_use_db, char **fqdn_out) {
+  void *vinfo;
+  const char *dsn = NULL, *fqdn = NULL;
+  int found = 0;
+  storagenode_info *info = NULL;
+  pthread_mutex_lock(&fqdn_to_info_cache_lock);
+  if(noit_hash_retrieve(&fqdn_to_info_cache, (void *)&id, sizeof(id),
+                        &vinfo)) {
+    found = 1;
+    info = vinfo;
+  }
+  pthread_mutex_unlock(&fqdn_to_info_cache_lock);
+  if(found) {
+    if(fqdn_out) *fqdn_out = info->fqdn;
+    return info->dsn;
+  }
+
+  if(!found && can_use_db) {
+    ds_single_detail *d;
+    conn_q *cq;
+    int row_count;
+    /* Look it up and store it */
+    d = calloc(1, sizeof(*d));
+    cq = get_conn_q_for_metanode();
+    GET_QUERY(find_storage);
+    DECLARE_PARAM_INT(id);
+    PG_EXEC(find_storage);
+    row_count = PQntuples(d->res);
+    if(row_count) {
+      PG_GET_STR_COL(dsn, 0, "dsn");
+      PG_GET_STR_COL(fqdn, 0, "fqdn");
+    }
+    PQclear(d->res);
+   bad_row:
+    free_params(d);
+    free(d);
+  }
+  if(fqdn) {
+    info = calloc(1, sizeof(*info));
+    info->fqdn = strdup(fqdn);
+    if(fqdn_out) *fqdn_out = info->fqdn;
+    info->dsn = dsn ? strdup(dsn) : NULL;
+    info->storagenode_id = id;
+    pthread_mutex_lock(&fqdn_to_info_cache_lock);
+    noit_hash_store(&fqdn_to_info_cache,
+                    (void *)&info->storagenode_id, sizeof(int), info);
+    pthread_mutex_unlock(&fqdn_to_info_cache_lock);
+  }
+  return info ? info->dsn : NULL;
+}
+static ds_line_detail *
+build_insert_batch(interim_journal_t *ij) {
+  int rv;
+  off_t len;
+  const char *buff, *cp, *lcp;
+  struct stat st;
+  ds_line_detail *head = NULL, *last = NULL, *next = NULL;
+
+  while((rv = fstat(ij->fd, &st)) == -1 && errno == EINTR);
+  assert(rv != -1);
+  len = st.st_size;
+  buff = mmap(NULL, len, PROT_READ, MAP_PRIVATE, ij->fd, 0);
+  if(buff == (void *)-1) {
+    noitL(noit_error, "mmap(%s) => %s\n", ij->filename, strerror(errno));
+    assert(buff != (void *)-1);
+  }
+  lcp = buff;
+  while(lcp < (buff + len) &&
+        NULL != (cp = strnstrn("\n", 1, lcp, len - (lcp-buff)))) {
+    next = calloc(1, sizeof(*next));
+    next->data = malloc(cp - lcp + 1);
+    memcpy(next->data, lcp, cp - lcp);
+    next->data[cp - lcp] = '\0';
+    if(!head) head = next;
+    if(last) last->next = next;
+    last = next;
+    lcp = cp + 1;
+  }
+  munmap((void *)buff, len);
+  return head;
+}
+static void
+interim_journal_remove(interim_journal_t *ij) {
+  unlink(ij->filename);
+  if(ij->filename) free(ij->filename);
+  if(ij->remote_str) free(ij->remote_str);
+  if(ij->remote_cn) free(ij->remote_cn);
+  if(ij->fqdn) free(ij->fqdn);
-}
-int
@@ -654 +974 @@
-                                  struct timeval *now) {
-  int i;
-
-
+
+
+
+
-  if(!(mask & EVENTER_ASYNCH_WORK)) return 0;
-  if(mask & EVENTER_ASYNCH_CLEANUP) return 0;
-
-
+
+
+
+
+
+
+
- full_monty:
-  /* Make sure we have a connection */
@@ -670 +997 @@
-  }
-
-
+
+
-  last_sp = NULL;
-  if(stratcon_datastore_do(cq, "BEGIN")) BUSTED(cq);
@@ -679 +1007 @@
- 
-      if(current->problematic) {
-noit_error, "[%s] Failed noit line: %s", cq->jobq->queue_name
+ingest_err, "%d\t%s\n", ij->storagenode_id
-        RELEASE_SAVEPOINT("batch");
-        current = current->next;
-        continue;
-      } 
-
+_str
-      switch(rv) {
-        case DS_EXEC_SUCCESS:
@@ -699 +1027 @@
-      }
-    }
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-  return 0;
+}
+static int
+stratcon_datastore_journal_sync(eventer_t e, int mask, void *closure,
+                                struct timeval *now) {
+  noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
+  const char *k;
+  int klen;
+  void *vij;
+  interim_journal_t *ij;
+  syncset_t *syncset = closure;
+
+  if(!(mask & EVENTER_ASYNCH_WORK)) return 0;
+  if(mask & EVENTER_ASYNCH_CLEANUP) return 0;
+
+  noitL(noit_debug, "Syncing journal sets...\n");
+  while(noit_hash_next(syncset->ws, &iter, &k, &klen, &vij)) {
+    eventer_t ingest;
+    ij = vij;
+    noitL(noit_debug, "Syncing journal set [%s,%s,%s]\n",
+          ij->remote_str, ij->remote_cn, ij->fqdn);
+    fsync(ij->fd);
+    ingest = eventer_alloc();
+    ingest->mask = EVENTER_ASYNCH;
+    ingest->callback = stratcon_datastore_asynch_execute;
+    ingest->closure = ij;
+    eventer_add_asynch(ij->cpool->jobq, ingest);
+  }
+  noit_hash_destroy(syncset->ws, free, NULL);
+  free(syncset->ws);
+  eventer_add(syncset->completion);
+  free(syncset);
+  return 0;
+}
+static interim_journal_t *
+interim_journal_get(struct sockaddr *remote, const char *remote_cn_in,
+                    int storagenode_id, const char *fqdn_in) {
+  void *vhash, *vij;
+  noit_hash_table *working_set;
+  interim_journal_t *ij;
+  struct timeval now;
+  char jpath[PATH_MAX];
+  char remote_str[128];
+  const char *remote_cn = remote_cn_in ? remote_cn_in : "default";
+  const char *fqdn = fqdn_in ? fqdn_in : "default";
+
+  convert_sockaddr_to_buff(remote_str, sizeof(remote_str), remote);
+  if(!*remote_str) strlcpy(remote_str, "default", sizeof(remote_str));
+
+  /* Lookup the working set */
+  if(!noit_hash_retrieve(&working_sets, remote_cn, strlen(remote_cn), &vhash)) {
+    working_set = calloc(1, sizeof(*working_set));
+    noit_hash_store(&working_sets, strdup(remote_cn), strlen(remote_cn),
+                    working_set);
+  }
+  else
+    working_set = vhash;
+
+  /* Lookup the interim journal within the working set */
+  if(!noit_hash_retrieve(working_set, fqdn, strlen(fqdn), &vij)) {
+    ij = calloc(1, sizeof(*ij));
+    gettimeofday(&now, NULL);
+    snprintf(jpath, sizeof(jpath), "%s/%s/%s/%d/%08x%08x",
+             basejpath, remote_str, remote_cn, storagenode_id,
+             (unsigned int)now.tv_sec, (unsigned int)now.tv_usec);
+    ij->remote_str = strdup(remote_str);
+    ij->remote_cn = strdup(remote_cn);
+    ij->fqdn = strdup(fqdn);
+    ij->storagenode_id = storagenode_id;
+    ij->filename = strdup(jpath);
+    ij->cpool = get_conn_pool_for_remote(ij->remote_str, ij->remote_cn,
+                                         ij->fqdn);
+    ij->fd = open(ij->filename, O_RDWR | O_CREAT | O_EXCL, 0640);
+    if(ij->fd < 0 && errno == ENOENT) {
+      if(mkdir_for_file(ij->filename, 0750)) {
+        noitL(noit_error, "Failed to create dir for '%s': %s\n",
+              ij->filename, strerror(errno));
+        exit(-1);
+      }
+      ij->fd = open(ij->filename, O_RDWR | O_CREAT | O_EXCL, 0640);
+    }
+    if(ij->fd < 0) {
+      noitL(noit_error, "Failed to open interim journal '%s': %s\n",
+            ij->filename, strerror(errno));
+      exit(-1);
+    }
+    noit_hash_store(working_set, strdup(fqdn), strlen(fqdn), ij);
+  }
+  else
+    ij = vij;
+
+  return ij;
+}
+static void
+storage_node_quick_lookup(const char *uuid_str, const char *remote_cn,
+                          int *sid_out, int *storagenode_id_out,
+                          char **fqdn_out, char **dsn_out) {
+  /* only called from the main thread -- no safety issues */
+  void *vuuidinfo, *vinfo;
+  uuid_info *uuidinfo;
+  storagenode_info *info = NULL;
+  char *fqdn = NULL;
+  char *dsn = NULL;
+  int storagenode_id = 0, sid = 0;
+  if(!noit_hash_retrieve(&uuid_to_info_cache, uuid_str, strlen(uuid_str),
+                         &vuuidinfo)) {
+    int row_count;
+    char *tmpint;
+    ds_single_detail *d;
+    conn_q *cq;
+    d = calloc(1, sizeof(*d));
+    cq = get_conn_q_for_metanode();
+    if(stratcon_database_connect(cq) == 0) {
+      /* Blocking call to service the cache miss */
+      GET_QUERY(check_map);
+      DECLARE_PARAM_STR(uuid_str, strlen(uuid_str));
+      DECLARE_PARAM_STR(remote_cn, strlen(remote_cn));
+      PG_EXEC(check_map);
+      row_count = PQntuples(d->res);
+      if(row_count != 1) {
+        PQclear(d->res);
+        goto bad_row;
+      }
+      PG_GET_STR_COL(tmpint, 0, "sid");
+      sid = atoi(tmpint);
+      PG_GET_STR_COL(tmpint, 0, "storage_node_id");
+      if(tmpint) storagenode_id = atoi(tmpint);
+      PG_GET_STR_COL(fqdn, 0, "fqdn");
+      PG_GET_STR_COL(dsn, 0, "dsn");
+      PQclear(d->res);
+    }
+   bad_row:
+    free_params((ds_single_detail *)d);
+    free(d);
+    release_conn_q(cq);
+    /* Place in cache */
+    if(fqdn) fqdn = strdup(fqdn);
+    uuidinfo = calloc(1, sizeof(*uuidinfo));
+    uuidinfo->sid = sid;
+    uuidinfo->uuid_str = strdup(uuid_str);
+    noit_hash_store(&uuid_to_info_cache,
+                    uuidinfo->uuid_str, strlen(uuidinfo->uuid_str), uuidinfo);
+    /* Also, we may have just witnessed a new storage node, store it */
+    if(storagenode_id) {
+      int needs_free = 0;
+      info = calloc(1, sizeof(*info));
+      info->storagenode_id = storagenode_id;
+      info->dsn = dsn ? strdup(dsn) : NULL;
+      info->fqdn = fqdn ? strdup(fqdn) : NULL;
+      pthread_mutex_lock(&fqdn_to_info_cache_lock);
+      if(!noit_hash_retrieve(&fqdn_to_info_cache,
+                             (void *)&storagenode_id, sizeof(int), &vinfo)) {
+        /* hack to save memory -- we *never* remove from these caches,
+           so we can use the same fqdn value in the above cache for the key
+           in the cache below -- (no strdup) */
+        noit_hash_store(&fqdn_to_info_cache,
+                        (void *)&info->storagenode_id, sizeof(int), info);
+      }
+      else needs_free = 1;
+      pthread_mutex_unlock(&fqdn_to_info_cache_lock);
+      if(needs_free) {
+        if(info->dsn) free(info->dsn);
+        if(info->fqdn) free(info->fqdn);
+        free(info);
+      }
+    }
+  }
+  else
+    uuidinfo = vuuidinfo;
+
+  if(storagenode_id) {
+    if(uuidinfo &&
+       ((!dsn && dsn_out) || (!fqdn && fqdn_out))) {
+      /* we don't have dsn and we actually want it */
+      pthread_mutex_lock(&fqdn_to_info_cache_lock);
+      if(noit_hash_retrieve(&fqdn_to_info_cache,
+                            (void *)&storagenode_id, sizeof(int), &vinfo))
+        info = vinfo;
+      pthread_mutex_unlock(&fqdn_to_info_cache_lock);
+    }
+  }
+
+  if(fqdn_out) *fqdn_out = fqdn ? fqdn : (info ? info->fqdn : NULL);
+  if(dsn_out) *dsn_out = dsn ? dsn : (info ? info->dsn : NULL);
+  if(storagenode_id_out) *storagenode_id_out = uuidinfo->storagenode_id;
+  if(sid_out) *sid_out = uuidinfo->sid;
+}
+static void
+stratcon_datastore_journal(struct sockaddr *remote,
+                           const char *remote_cn, const char *line) {
+  interim_journal_t *ij = NULL;
+  char uuid_str[UUID_STR_LEN+1], *cp, *fqdn, *dsn;
+  int storagenode_id = 0;
+  uuid_t checkid;
+  if(!line) return;
+  /* if it is a UUID based thing, find the storage node */
+  switch(*line) {
+    case 'C':
+    case 'S':
+    case 'M':
+      if(line[1] == '\t' && (cp = strchr(line+2, '\t')) != NULL) {
+        strlcpy(uuid_str, cp + 1, sizeof(uuid_str));
+        if(!uuid_parse(uuid_str, checkid)) {
+          storage_node_quick_lookup(uuid_str, remote_cn, NULL,
+                                    &storagenode_id, &fqdn, &dsn);
+          ij = interim_journal_get(remote, remote_cn, storagenode_id, fqdn);
+        }
+      }
+      break;
+    case 'n':
+      ij = interim_journal_get(remote,remote_cn,0,NULL);
+      break;
+    default:
+      break;
+  }
+  if(!ij && fqdn) {
+    noitL(ingest_err, "%d\t%s\n", storagenode_id, line);
+  }
+  else {
+    int len;
+    len = write(ij->fd, line, strlen(line));
+    if(len < 0) {
+      noitL(noit_error, "write to %s failed: %s\n",
+            ij->filename, strerror(errno));
+    }
+  }
+  return;
+}
+static noit_hash_table *
+stratcon_datastore_journal_remove(struct sockaddr *remote,
+                                  const char *remote_cn) {
+  void *vhash = NULL;
+  if(noit_hash_retrieve(&working_sets, remote_cn, strlen(remote_cn), &vhash)) {
+    /* pluck it out */
+    noit_hash_delete(&working_sets, remote_cn, strlen(remote_cn), free, NULL);
+  }
+  else {
+    noitL(noit_error, "attempted checkpoint on non-existing workingset: '%s'\n",
+          remote_cn);
+    abort();
+  }
+  return vhash;
-}
-void
-stratcon_datastore_push(stratcon_datastore_op_t op,
-
+
+
+
-  conn_q *cq;
+  syncset_t *syncset;
-  eventer_t e;
-job_detail *dsjd
+rt_detail *rtdetail
-  struct datastore_onlooker_list *nnode;
-
-  for(nnode = onlookers; nnode; nnode = nnode->next)
-
-
-
-
+
+
-  switch(op) {
-
-
-
+
+
-      break;
-
-
-
+
+
+
+
+
+
+
+
+
-      break;
-    case DS_OP_FIND_COMPLETE:
-
-
-
+
+
+
+
-      e = eventer_alloc();
-      e->mask = EVENTER_ASYNCH;
-
-
-
-
-
-
+
+
+
-      break;
-  }
@@ -750 +1329 @@
-stratcon_datastore_saveconfig(void *unused) {
-  int rv = -1;
-  conn_q _cq = { 0 }, *cq = &_cq;
-  char *buff;
-  ds_single_detail _d = { 0 }, *d = &_d;
+  conn_q *cq;
+  cq = get_conn_q_for_metanode();
-
-  if(stratcon_database_connect(cq) == 0) {
@@ -775 +1355 @@
-      free_params(d);
-  }
-if(cq->dbh) PQfinish(cq->dbh
+release_conn_q(cq
-  return rv;
-}
@@ -781 +1361 @@
-void
-stratcon_datastore_register_onlooker(void (*f)(stratcon_datastore_op_t,
-
+
+
-  struct datastore_onlooker_list *nnode;
-  nnode = calloc(1, sizeof(*nnode));
@@ -789 +1370 @@
-    nnode->next = onlookers;
-}
+static void
+stratcon_datastore_launch_file_ingestion(char *remote_str, char *remote_cn,
+                                         char *id_str, char *file) {
+  char path[PATH_MAX];
+  interim_journal_t *ij;
+  eventer_t ingest;
+
+  snprintf(path, sizeof(path), "%s/%s/%s/%s/%s",
+           basejpath, remote_str, remote_cn, id_str, file);
+  ij = calloc(1, sizeof(*ij));
+  ij->fd = open(path, O_RDONLY);
+  if(ij->fd < 0) {
+    noitL(noit_error, "cannot open journal '%s': %s\n",
+          path, strerror(errno));
+    free(ij);
+    return;
+  }
+  ij->filename = strdup(path);
+  ij->remote_str = strdup(remote_str);
+  ij->remote_cn = strdup(remote_cn);
+  ij->storagenode_id = atoi(id_str);
+  ij->cpool = get_conn_pool_for_remote(ij->remote_str, ij->remote_cn,
+                                       ij->fqdn);
+
+  noitL(noit_error, "ingesting old payload: %s\n", ij->filename);
+  ingest = eventer_alloc();
+  ingest->mask = EVENTER_ASYNCH;
+  ingest->callback = stratcon_datastore_asynch_execute;
+  ingest->closure = ij;
+  eventer_add_asynch(ij->cpool->jobq, ingest);
+}
+static void
+stratcon_datastore_sweep_journals_int(char *first, char *second, char *third) {
+  char path[PATH_MAX];
+  DIR *root;
+  struct dirent de, *entry;
+  int i = 0, cnt = 0;
+  char **entries;
+
+  snprintf(path, sizeof(path), "%s%s%s%s%s%s%s", basejpath,
+           first ? "/" : "", first ? first : "",
+           second ? "/" : "", second ? second : "",
+           third ? "/" : "", third ? third : "");
+  root = opendir(path);
+  if(!root) return;
+  while(readdir_r(root, &de, &entry) == 0 && entry != NULL) cnt++;
+  rewinddir(root);
+  entries = malloc(sizeof(*entries) * cnt);
+  while(readdir_r(root, &de, &entry) == 0 && entry != NULL) {
+    if(i < cnt) {
+      entries[i++] = strdup(entry->d_name);
+    }
+  }
+  closedir(root);
+  cnt = i; /* could have changed, directories are fickle */
+  qsort(entries, i, sizeof(*entries),
+        (int (*)(const void *, const void *))strcasecmp);
+  for(i=0; i<cnt; i++) {
+    if(!strcmp(entries[i], ".") || !strcmp(entries[i], "..")) continue;
+    if(!first)
+      stratcon_datastore_sweep_journals_int(entries[i], NULL, NULL);
+    else if(!second)
+      stratcon_datastore_sweep_journals_int(first, entries[i], NULL);
+    else if(!third)
+      stratcon_datastore_sweep_journals_int(first, second, entries[i]);
+    else if(strlen(entries[i]) == 16)
+      stratcon_datastore_launch_file_ingestion(first,second,third,entries[i]);
+  }
+}
+static void
+stratcon_datastore_sweep_journals() {
+  stratcon_datastore_sweep_journals_int(NULL,NULL,NULL);
+}
+
+int
+stratcon_datastore_ingest_all_storagenode_info() {
+  int i, cnt;
+  ds_single_detail _d = { 0 }, *d = &_d;
+  conn_q *cq;
+  cq = get_conn_q_for_metanode();
+
+  while(stratcon_database_connect(cq)) {
+    noitL(noit_error, "Error connecting to database\n");
+    sleep(1);
+  }
+
+  GET_QUERY(all_storage);
+  PG_EXEC(all_storage);
+  cnt = PQntuples(d->res);
+  for(i=0; i<cnt; i++) {
+    void *vinfo;
+    char *tmpint, *fqdn, *dsn;
+    int storagenode_id;
+    PG_GET_STR_COL(tmpint, i, "storage_node_id");
+    storagenode_id = atoi(tmpint);
+    PG_GET_STR_COL(fqdn, i, "fqdn");
+    PG_GET_STR_COL(dsn, i, "dsn");
+    PG_GET_STR_COL(tmpint, i, "storage_node_id");
+    storagenode_id = tmpint ? atoi(tmpint) : 0;
+
+    if(!noit_hash_retrieve(&fqdn_to_info_cache,
+                           (void *)&storagenode_id, sizeof(int), &vinfo)) {
+      storagenode_info *info;
+      info = calloc(1, sizeof(*info));
+      info->storagenode_id = storagenode_id;
+      info->fqdn = fqdn ? strdup(fqdn) : NULL;
+      info->dsn = dsn ? strdup(dsn) : NULL;
+      noit_hash_store(&fqdn_to_info_cache,
+                      (void *)&info->storagenode_id, sizeof(int), info);
+    }
+  }
+  PQclear(d->res);
+ bad_row:
+  free_params(d);
+
+  release_conn_q(cq);
+  noitL(noit_error, "Loaded %d storage nodes\n", cnt);
+  return cnt;
+}
+int
+stratcon_datastore_ingest_all_check_info() {
+  int i, cnt, loaded = 0;
+  ds_single_detail _d = { 0 }, *d = &_d;
+  conn_q *cq;
+  cq = get_conn_q_for_metanode();
+
+  while(stratcon_database_connect(cq)) {
+    noitL(noit_error, "Error connecting to database\n");
+    sleep(1);
+  }
+
+  GET_QUERY(check_mapall);
+  PG_EXEC(check_mapall);
+  cnt = PQntuples(d->res);
+  for(i=0; i<cnt; i++) {
+    void *vinfo;
+    char *tmpint, *fqdn, *dsn, *uuid_str;
+    int sid, storagenode_id;
+    uuid_info *uuidinfo;
+    PG_GET_STR_COL(uuid_str, i, "id");
+    if(!uuid_str) continue;
+    PG_GET_STR_COL(tmpint, i, "sid");
+    if(!tmpint) continue;
+    sid = atoi(tmpint);
+    PG_GET_STR_COL(fqdn, i, "fqdn");
+    PG_GET_STR_COL(dsn, i, "dsn");
+    PG_GET_STR_COL(tmpint, i, "storage_node_id");
+    storagenode_id = tmpint ? atoi(tmpint) : 0;
+
+    uuidinfo = calloc(1, sizeof(*uuidinfo));
+    uuidinfo->uuid_str = strdup(uuid_str);
+    uuidinfo->sid = sid;
+    noit_hash_store(&uuid_to_info_cache,
+                    uuidinfo->uuid_str, strlen(uuidinfo->uuid_str), uuidinfo);
+    loaded++;
+    if(!noit_hash_retrieve(&fqdn_to_info_cache,
+                           (void *)&storagenode_id, sizeof(int), &vinfo)) {
+      storagenode_info *info;
+      info = calloc(1, sizeof(*info));
+      info->storagenode_id = storagenode_id;
+      info->fqdn = fqdn ? strdup(fqdn) : NULL;
+      info->dsn = dsn ? strdup(dsn) : NULL;
+      noit_hash_store(&fqdn_to_info_cache,
+                      (void *)&info->storagenode_id, sizeof(int), info);
+    }
+  }
+  PQclear(d->res);
+ bad_row:
+  free_params(d);
+
+  release_conn_q(cq);
+  noitL(noit_error, "Loaded %d uuid -> (sid,storage_node_id) mappings\n", loaded);
+  return loaded;
+}
+void
+stratcon_datastore_init() {
+  pthread_mutex_init(&ds_conns_lock, NULL);
+  pthread_mutex_init(&fqdn_to_info_cache_lock, NULL);
+  ds_err = noit_log_stream_find("error/datastore");
+  ingest_err = noit_log_stream_find("error/ingest");
+  if(!ds_err) ds_err = noit_error;
+  if(!ingest_err) ingest_err = noit_error;
+  if(!noit_conf_get_string(NULL, "/stratcon/database/journal/path",
+                           &basejpath)) {
+    noitL(noit_error, "/stratcon/database/journal/path is unspecified\n");
+    exit(-1);
+  }
+  stratcon_datastore_ingest_all_check_info();
+  stratcon_datastore_ingest_all_storagenode_info();
+  stratcon_datastore_sweep_journals();
+}

src/stratcon_datastore.h

@@ -44 +44 @@
- DS_OP_INSERT = 1,
- DS_OP_CHKPT = 2,
-
-
+
-} stratcon_datastore_op_t;
-
-API_EXPORT(void)
-  stratcon_datastore_push(stratcon_datastore_op_t,
-void *
+const char *, void *, eventer_t
-
-API_EXPORT(void)
-  stratcon_datastore_register_onlooker(void (*f)(stratcon_datastore_op_t,
-
+
+
+
+
+
-
-API_EXPORT(int)

src/stratcon_iep.c

@@ -142 +142 @@
-  }
-  noitL(noit_error, "submitting statement: %s\n", line);
-line
+NULL, line, NULL
-  stmt->marked = 1;
-}
@@ -163 +163 @@
-  /* Phase 1: sweep in all the statements */
-  for(i=0; i<cnt; i++) {
-
++1
-    char provides[256];
-    char *statement;
@@ -202 +202 @@
-  /* Phase 2: load the requires graph */
-  for(i=0; i<cnt; i++) {
-
++1
-    int rcnt, j;
-    char *requires;
@@ -280 +280 @@
-  noitL(noit_debug, "Found %d %s stanzas\n", cnt, path);
-  for(i=0; i<cnt; i++) {
-
++1
-    char topic[256];
-    char *query;
@@ -314 +314 @@
-      query++;
-    }
-line
+NULL, line, NULL
-  }
-  free(query_configs);
@@ -419 +419 @@
-setup_iep_connection_callback(eventer_t e, int mask, void *closure,
-                              struct timeval *now) {
-
+, NULL, NULL
-  return 0;
-}
@@ -526 +526 @@
-void
-stratcon_iep_line_processor(stratcon_datastore_op_t op,
-
+
+
-  int len;
-  char remote_str[128];

src/stratcon_iep.h

@@ -48 +48 @@
-API_EXPORT(void)
-  stratcon_iep_line_processor(stratcon_datastore_op_t op,
-
+
+
-
-API_EXPORT(jlog_streamer_ctx_t *)

src/stratcon_jlog_streamer.c

@@ -371 +371 @@
-        FULLREAD(e, ctx, (unsigned long)ctx->header.message_len);
-        if(ctx->header.message_len > 0)
-
+
+
-        else if(ctx->buffer)
-          free(ctx->buffer);
@@ -385 +386 @@
-          completion_e->mask = EVENTER_READ | EVENTER_WRITE | EVENTER_EXCEPTION;
-          ctx->state = JLOG_STREAMER_IS_ASYNC;
-
+
+
-          noitL(noit_debug, "Pushing batch asynch...\n");
-          return 0;

src/stratcon_jlog_streamer.h

@@ -91 +91 @@
-  u_int64_t total_bytes_read;
-
-void *
+const char *, void *, eventer_t
-} jlog_streamer_ctx_t;
-

src/stratcon_realtime_http.c

@@ -427 +427 @@
-    for(node = rc->checklist; node; node = node->next) {
-      noit_atomic_inc32(&ctx->ref_cnt);
-      stratcon_datastore_push(DS_OP_FIND, NULL, node);
-      noitL(noit_error, "Resolving sid: %d\n", node->sid);
-    }
@@ -435 +434 @@
-    completion->closure = ctx;
-    gettimeofday(&completion->whence, NULL);
-
+
+
-  }
-  return EVENTER_EXCEPTION;

src/stratcond.c

@@ -171 +171 @@
-  noit_watchdog_child_eventer_heartbeat();
-
+  stratcon_datastore_init();
-  noit_console_init(APPNAME);
-  noit_console_conf_init();