Changeset 9799031c4d6cbb100f1ccbc5c773a07e3d988542 for src

Timestamp:

09/07/11 18:43:47 (2 years ago)

Author:

Theo Schlossnagle <jesus@omniti.com>

git-committer:

Theo Schlossnagle <jesus@omniti.com> 1315421027 -0400

git-parent:

[c6d83bbfb23e3a87bbd661cc601af89380951a49]

git-author:

Theo Schlossnagle <jesus@omniti.com> 1315421027 -0400

Message:

support multiple ingestors

Files:

• src/modules/handoff_ingestor.c (modified) (2 diffs)
• src/modules/postgres_ingestor.c (modified) (5 diffs)
• src/stratcon_datastore.c (modified) (5 diffs)
• src/stratcon_datastore.h (modified) (3 diffs)

src/modules/handoff_ingestor.c

@@ -175 +175 @@
-}
-
-void
+int
-stratcon_ingest_launch_file_ingestion(const char *path,
-                                      const char *remote_str,
-                                      const char *remote_cn,
-                                      const char *id_str) {
-
-
+
+
+
+
+
+
+
+
+
+
+
+
-  if(the_one_and_only) {
-    noit_http_session_ctx *ctx = the_one_and_only;
-path
+hfile
-    if(noit_http_response_append(ctx,msg,strlen(msg)) == noit_false ||
-       noit_http_response_flush(ctx, noit_false) == noit_false) {
@@ -191 +201 @@
-    }
-  }
+  return 0;
-}
-

src/modules/postgres_ingestor.c

@@ -1413 +1413 @@
-}
-
-void
+int
-stratcon_ingest_launch_file_ingestion(const char *path,
-                                      const char *remote_str,
@@ -1419 +1419 @@
-                                      const char *id_str) {
-  pg_interim_journal_t *ij;
+  char pgfile[PATH_MAX];
-  eventer_t ingest;
-
+  if(strcmp(path + strlen(path) - 3, ".pg")) {
+    snprintf(pgfile, sizeof(pgfile), "%s.pg", path);
+    if(link(path, pgfile) < 0) {
+      noitL(noit_error, "cannot link journal: %s\n", strerror(errno));
+      free(ij);
+      return -1;
+    }
+  }
+  else
+    strlcpy(pgfile, path, sizeof(pgfile));
-  ij = calloc(1, sizeof(*ij));
-ath
+gfile
-  if(ij->fd < 0) {
-    noitL(noit_error, "cannot open journal '%s': %s\n",
-ath
+gfile
-    free(ij);
-
+ -1
-  }
-  close(ij->fd);
-  ij->fd = -1;
-ath
+gfile
-  ij->remote_str = strdup(remote_str);
-  ij->remote_cn = strdup(remote_cn);
@@ -1443 +1454 @@
-  ingest->closure = ij;
-  eventer_add_asynch(ij->cpool->jobq, ingest);
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
-}
-
@@ -1680 +1633 @@
-  return 0;
-}
+static int is_postgres_ingestor_file(const char *file) {
+  return (strlen(file) == 19 && !strcmp(file + 16, ".pg"));
+}
-static int postgres_ingestor_init(noit_module_generic_t *self) {
-  pthread_mutex_init(&ds_conns_lock, NULL);
@@ -1696 +1652 @@
-  stratcon_ingest_all_check_info();
-  stratcon_ingest_all_storagenode_info();
-
+
+
-  return stratcon_datastore_set_ingestor(&postgres_ingestor_api);
-}

src/stratcon_datastore.c

@@ -62 +62 @@
-
-static ingestor_api_t *ingestor = NULL;
+typedef struct ingest_chain_t {
+  ingestor_api_t *ingestor;
+  struct ingest_chain_t *next;
+} ingest_chain_t;
+
+static ingest_chain_t *ingestor_chain;
+
-static int ds_system_enabled = 1;
-int stratcon_datastore_get_enabled() { return ds_system_enabled; }
-void stratcon_datastore_set_enabled(int n) { ds_system_enabled = n; }
-int stratcon_datastore_set_ingestor(ingestor_api_t *ni) {
-
-
+
+
+
+
+
+
+
+
-  return 0;
-}
@@ -92 +105 @@
-  if(ij->fqdn) free(ij->fqdn);
-  free(ij);
+}
+
+static void
+stratcon_ingest_sweep_journals_int(char *first, char *second, char *third,
+                                   int (*test)(const char *),
+                                   int (*ingest)(const char *fullpath,
+                                                 const char *remote_str,
+                                                 const char *remote_cn,
+                                                 const char *id_str)) {
+  char path[PATH_MAX];
+  DIR *root;
+  struct dirent *de, *entry;
+  int i = 0, cnt = 0;
+  char **entries;
+  int size = 0;
+
+  snprintf(path, sizeof(path), "%s%s%s%s%s%s%s", basejpath,
+           first ? "/" : "", first ? first : "",
+           second ? "/" : "", second ? second : "",
+           third ? "/" : "", third ? third : "");
+#ifdef _PC_NAME_MAX
+  size = pathconf(path, _PC_NAME_MAX);
+#endif
+  size = MAX(size, PATH_MAX + 128);
+  de = alloca(size);
+  root = opendir(path);
+  if(!root) return;
+  while(portable_readdir_r(root, de, &entry) == 0 && entry != NULL) cnt++;
+  closedir(root);
+  root = opendir(path);
+  if(!root) return;
+  entries = malloc(sizeof(*entries) * cnt);
+  while(portable_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;
+    noitL(ds_deb, "Processing L%d entry '%s'\n",
+          third ? 4 : second ? 3 : first ? 2 : 1, entries[i]);
+    if(!first)
+      stratcon_ingest_sweep_journals_int(entries[i], NULL, NULL, test, ingest);
+    else if(!second)
+      stratcon_ingest_sweep_journals_int(first, entries[i], NULL, test, ingest);
+    else if(!third)
+      stratcon_ingest_sweep_journals_int(first, second, entries[i], test, ingest);
+    else if(test(entries[i])) {
+      char fullpath[PATH_MAX];
+      snprintf(fullpath, sizeof(fullpath), "%s/%s/%s/%s/%s", basejpath,
+               first,second,third,entries[i]);
+      ingest(fullpath,first,second,third);
+    }
+  }
+  for(i=0; i<cnt; i++)
+    free(entries[i]);
+  free(entries);
+}
+void
+stratcon_ingest_sweep_journals(int (*test)(const char *),
+                               int (*ingest)(const char *fullpath,
+                                             const char *remote_str,
+                                             const char *remote_cn,
+                                             const char *id_str)) {
+  stratcon_ingest_sweep_journals_int(NULL,NULL,NULL, test, ingest);
+}
+
+static int
+stratcon_ingest(const char *fullpath, const char *remote_str,
+                const char *remote_cn, const char *id_str) {
+  ingest_chain_t *ic;
+  int err = 0;
+  for(ic = ingestor_chain; ic; ic = ic->next)
+    if(ic->ingestor->launch_file_ingestion(fullpath, remote_str,
+                                           remote_cn, id_str))
+      err = -1;
+  if(err == 0) {
+    unlink(fullpath);
+  }
+  return err;
-}
-static int
@@ -139 +236 @@
-    ij->fd = -1;
-    snprintf(id_str, sizeof(id_str), "%d", ij->storagenode_id);
-ingestor->launch_file_ingestion
-                
+stratcon_ingest
+
-  }
-  noit_hash_destroy(syncset->ws, free, interim_journal_free);
@@ -368 +465 @@
-}
-
-
-
+
+
+
+
+
+
-  ds_err = noit_log_stream_find("error/datastore");
-  ds_deb = noit_log_stream_find("debug/datastore");
@@ -381 +482 @@
-    exit(-1);
-  }
+}
+void
+stratcon_datastore_init() {
+  stratcon_datastore_core_init();
+
+  stratcon_ingest_sweep_journals(is_raw_ingestion_file,
+                                 stratcon_ingest);
-
-  assert(noit_http_rest_register_auth(

src/stratcon_datastore.h

@@ -43 +43 @@
-
-typedef struct {
-void
- 
+int
+
-  void (*iep_check_preload)();
-  int (*storage_node_lookup)(const char *uuid_str, const char *remote_cn,
@@ -83 +83 @@
-
-API_EXPORT(void)
+  stratcon_datastore_core_init();
+
+API_EXPORT(void)
-  stratcon_datastore_init();
-
@@ -98 +101 @@
-  stratcon_datastore_set_enabled(int);
-
+API_EXPORT(void)
+  stratcon_ingest_sweep_journals(int (*test)(const char *),
+                                 int (*ingest)(const char *fullpath,
+                                               const char *remote_str,
+                                               const char *remote_cn,
+                                               const char *id_str));
-#endif