Changeset 699c97c25d44d177d2979ba0008d46658927a912

Timestamp:

08/06/09 13:21:06 (4 years ago)

Author:

Theo Schlossnagle <jesus@omniti.com>

git-committer:

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

git-parent:

[e754a7ed9a2306bb7b477de6a099909fc3234c53]

git-author:

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

Message:

works for me. Ensures there is a (perhaps non-connected) DAG of statement dependencies and runs them, then loads data, then runs queries. refs #162

Files:

• src/java/Makefile.in (modified) (1 diff)
• src/java/com/omniti/reconnoiter/EventHandler.java (modified) (3 diffs)
• src/java/com/omniti/reconnoiter/IEPEngine.java (modified) (1 diff)
• src/java/com/omniti/reconnoiter/MQListener.java (modified) (1 diff)
• src/java/com/omniti/reconnoiter/StratconMessage.java (modified) (1 diff)
• src/java/com/omniti/reconnoiter/event/StratconQuery.java (modified) (3 diffs)
• src/java/com/omniti/reconnoiter/event/StratconQueryBase.java (added)
• src/java/com/omniti/reconnoiter/event/StratconStatement.java (added)
• src/stratcon.conf.in (modified) (1 diff)
• src/stratcon_iep.c (modified) (5 diffs)

src/java/Makefile.in

@@ -32 +32 @@
-        com/omniti/reconnoiter/event/NoitEvent.java \
-        com/omniti/reconnoiter/event/NoitMetricNumeric.java \
+        com/omniti/reconnoiter/event/StratconStatement.java \
+        com/omniti/reconnoiter/event/StratconQueryBase.java \
-        com/omniti/reconnoiter/event/StratconQuery.java \
-        com/omniti/reconnoiter/event/StratconQueryStop.java \

src/java/com/omniti/reconnoiter/EventHandler.java

@@ -10 +10 @@
-import com.omniti.reconnoiter.event.NoitEvent;
-import com.omniti.reconnoiter.event.NoitMetricNumeric;
+import com.omniti.reconnoiter.event.StratconQueryBase;
+import com.omniti.reconnoiter.event.StratconStatement;
-import com.omniti.reconnoiter.event.StratconQuery;
-import com.omniti.reconnoiter.event.StratconQueryStop;
@@ -15 +17 @@
-public class EventHandler {
-        
-       
-       private ConcurrentHashMap<UUID, StratconQuery
-       
+  
+  private ConcurrentHashMap<UUID, StratconQueryBase
+  
-
-       public EventHandler(ConcurrentHashMap<UUID,StratconQuery
-               
-               
-               
-       
+  public EventHandler(ConcurrentHashMap<UUID,StratconQueryBase
+    
+    
+    
+  
-        
-
-
-
+
+
-    if(m == null) {
-      System.err.println("Can't grok:\n" + xml);
-    }
-
+
+
+
+
+
+
+
+
+
+
+
-      StratconQuery sq = (StratconQuery) m;
-
@@ -46 +57 @@
-    }
-    else if(m instanceof StratconQueryStop) {
-
+
+
-      if(sq != null) {
-        queries.remove(sq.getUUID());

src/java/com/omniti/reconnoiter/IEPEngine.java

@@ -34 +34 @@
-    NoitEvent.registerTypes(epService);
-
-    epService.getEPAdministrator().createEPL("create window CheckDetails.std:unique(uuid).win:keepall() as NoitCheck");
-    epService.getEPAdministrator().createEPL("insert into CheckDetails select * from NoitCheck");
-
-    MQListener l = new MQListener(epService, BrokerFactory.getBroker(sconf));
-

src/java/com/omniti/reconnoiter/MQListener.java

@@ -19 +19 @@
-public class MQListener implements Runnable {
-    private EPServiceProvider epService;
-
+Base
-    private IMQBroker broker;
-
-    public MQListener(EPServiceProvider epService, IMQBroker broker) {
-
+Base
-      this.epService = epService;
-      this.broker = broker;

src/java/com/omniti/reconnoiter/StratconMessage.java

@@ -59 +59 @@
-      }
-      // and requests
+      else if(tag.equals("StratconStatement"))
+        return new StratconStatement(document);
-      else if(tag.equals("StratconQuery"))
-        return new StratconQuery(document);

src/java/com/omniti/reconnoiter/event/StratconQuery.java

@@ -11 +11 @@
-import java.lang.System;
-
+import com.omniti.reconnoiter.event.StratconQueryBase;
-import com.omniti.reconnoiter.StratconMessage;
-import com.espertech.esper.client.EPStatement;
@@ -24 +25 @@
-import org.w3c.dom.NodeList;
-
-
-
-
-
-
-
+
+
+
-
-  public StratconQuery(Document d) {
@@ -45 +43 @@
-    if(uuid == null) uuid = UUID.randomUUID();
-  }
-  public UUID getUUID() {
-    return uuid;
-  }
-  public EPStatement getStatement() {
-    return statement;
-  }
-  public String getExpression() {
-    return expression;
-  }
-  public String getName() {
-    return name;
-  }
-  public void setStatement(EPStatement s) {
-    this.statement = s;
-  }
-  public void setListener(UpdateListener l) {

src/stratcon.conf.in

@@ -35 +35 @@
-           command="%iepbindir%/run-iep.sh" />
-    <queries>
+      <statement id="6cc613a4-7f9c-11de-973f-db7e8ccb2e5c" provides="CheckDetails-ddl">
+        <epl>create window CheckDetails.std:unique(uuid).win:keepall() as NoitCheck</epl>
+      </statement>
+      <statement id="76598f5e-7f9c-11de-9f5b-ebb4dcb2494e" provides="CheckDetails">
+        <requires>CheckDetails-ddl</requires>
+        <epl>insert into CheckDetails select * from NoitCheck</epl>
+      </statement>
+      <statement id="ba189f08-7f99-11de-9013-733772d37479" provides="UnavailableStream">
+        <requires>CheckDetails</requires>
+        <epl>insert into UnavailableStream
+             select p.* as delta, cds.target as target, cds.module as module,
+                    cds.name as name, p.s.uuid as uuid
+             from pattern [ every
+                            s=NoitStatus(availability='A') ->
+                            ( n0 = NoitStatus(uuid=s.uuid, availability='U')
+                              and not NoitStatus(uuid=s.uuid, availability='A'))
+                          ].std:lastevent() as p
+             inner join CheckDetails as cds on cds.uuid = p.s.uuid
+        </epl>
+      </statement>
-      <query id="ce6bf8d2-3dd7-11de-a45c-a7df160cba9e" topic="status">
-select * from NoitStatus
+<epl>select * from NoitStatus</epl>
-      </query>
-    </queries>

src/stratcon_iep.c

@@ -169 +169 @@
-
-static xmlDocPtr
+stratcon_iep_doc_from_statement(char *data, char *remote) {
+  xmlDocPtr doc;
+  char *parts[3];
+  if(bust_to_parts(data, parts, 3) != 3) return NULL;
+  /*  'D' ID QUERY  */
+
+  NEWDOC(doc, "StratconStatement",
+         {
+           ADDCHILD("id", parts[1]);
+           ADDCHILD("expression", parts[2]);
+         });
+  return doc;
+}
+
+static xmlDocPtr
-stratcon_iep_doc_from_query(char *data, char *remote) {
-  xmlDocPtr doc;
@@ -205 +220 @@
-      case 'S': return stratcon_iep_doc_from_status(data, remote);
-      case 'M': return stratcon_iep_doc_from_metric(data, remote);
+      case 'D': return stratcon_iep_doc_from_statement(data, remote);
-      case 'Q': return stratcon_iep_doc_from_query(data, remote);
-      case 'q': return stratcon_iep_doc_from_querystop(data, remote);
@@ -222 +238 @@
-  }
-  return 0;
+}
+
+struct statement_node {
+  char *id;
+  char *statement;
+  char *provides;
+  int marked; /* helps with identifying cycles */
+  int nrequires;
+  struct statement_node **requires;
+};
+static void
+statement_node_free(void *vstmt) {
+  struct statement_node *stmt = vstmt;
+  if(stmt->id) free(stmt->id);
+  if(stmt->statement) free(stmt->statement);
+  if(stmt->provides) free(stmt->provides);
+  if(stmt->requires) free(stmt->requires);
+}
+static int
+stmt_mark_dag(struct statement_node *stmt, int mgen) {
+  int i;
+  assert(stmt->marked <= mgen);
+  if(stmt->marked == mgen) return -1;
+  if(stmt->marked > 0) return 0; /* validated in a previous sweep */
+  stmt->marked = mgen;
+  for(i=0; i<stmt->nrequires; i++)
+    if(stmt_mark_dag(stmt->requires[i], mgen) < 0) return -1;
+  return 0;
+}
+static void
+submit_statement_node(struct statement_node *stmt) {
+  int line_len, i;
+  char *line, *cp;
+
+  if(stmt->marked) return;
+  for(i=0; i<stmt->nrequires; i++)
+    submit_statement_node(stmt->requires[i]);
+
+  line_len = 3 /* 2 tabs + \0 */ +
+             1 /* 'D' */ + 1 /* '\n' */ +
+             strlen(stmt->id) + strlen(stmt->statement);
+  line = malloc(line_len);
+  snprintf(line, line_len, "D\t%s\t%s\n", stmt->id, stmt->statement);
+  cp = line;
+  while(cp[0] && cp[1]) {
+    if(*cp == '\n') *cp = ' ';
+    cp++;
+  }
+  noitL(noit_error, "submitting statement: %s\n", line);
+  stratcon_iep_line_processor(DS_OP_INSERT, NULL, line);
+  stmt->marked = 1;
+}
+void stratcon_iep_submit_statements() {
+  int i, cnt = 0;
+  noit_conf_section_t *statement_configs;
+  char path[256];
+  struct statement_node *stmt;
+  void *vstmt;
+  noit_hash_table stmt_by_id = NOIT_HASH_EMPTY;
+  noit_hash_table stmt_by_provider = NOIT_HASH_EMPTY;
+  noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
+  const char *key;
+  int klen, mgen = 0;
+
+  snprintf(path, sizeof(path), "/stratcon/iep/queries//statement");
+  statement_configs = noit_conf_get_sections(NULL, path, &cnt);
+  noitL(noit_debug, "Found %d %s stanzas\n", cnt, path);
+
+  /* Phase 1: sweep in all the statements */
+  for(i=0; i<cnt; i++) {
+    char id[UUID_STR_LEN];
+    char provides[256];
+    char *statement;
+
+    if(!noit_conf_get_stringbuf(statement_configs[i],
+                                "self::node()/@id",
+                                id, sizeof(id))) {
+      noitL(noit_iep, "No uuid specified in query\n");
+      continue;
+    }
+    if(!noit_conf_get_stringbuf(statement_configs[i],
+                                "ancestor-or-self::node()/@provides",
+                                provides, sizeof(provides))) {
+      provides[0] = '\0';
+    }
+    if(!noit_conf_get_string(statement_configs[i], "self::node()/epl",
+                             &statement)) {
+      noitL(noit_iep, "No contents specified in statement\n");
+      continue;
+    }
+    stmt = calloc(1, sizeof(*stmt));
+    stmt->id = strdup(id);
+    stmt->statement = statement;
+    stmt->provides = provides[0] ? strdup(provides) : NULL;
+    if(!noit_hash_store(&stmt_by_id, stmt->id, strlen(stmt->id), stmt)) {
+      noitL(noit_error, "Duplicate statement id: %s\n", stmt->id);
+      exit(-1);
+    }
+    if(stmt->provides) {
+      if(!noit_hash_store(&stmt_by_provider, stmt->provides,
+                          strlen(stmt->provides), stmt)) {
+        noitL(noit_error, "Two statements provide: '%s'\n", stmt->provides);
+        exit(-1);
+      }
+    }
+  }
+
+  /* Phase 2: load the requires graph */
+  for(i=0; i<cnt; i++) {
+    char id[UUID_STR_LEN];
+    int rcnt, j;
+    char *requires;
+    noit_conf_section_t *reqs;
+
+    if(!noit_conf_get_stringbuf(statement_configs[i],
+                                "self::node()/@id",
+                                id, sizeof(id))) {
+      noitL(noit_iep, "No uuid specified in query\n");
+      continue;
+    }
+    if(!noit_hash_retrieve(&stmt_by_id, id, strlen(id), &vstmt)) {
+      noitL(noit_error, "Cannot find statement: %s\n", id);
+      exit(-1);
+    }
+    stmt = vstmt;
+    reqs = noit_conf_get_sections(statement_configs[i],
+                                  "self::node()/requires", &rcnt);
+    if(rcnt > 0) {
+      stmt->requires = malloc(rcnt * sizeof(*(stmt->requires)));
+      for(j=0; j<rcnt; j++) {
+        void *vrstmt;
+        if(!noit_conf_get_string(reqs[j], "self::node()",
+                                 &requires) || requires[0] == '\0') {
+          continue;
+        }
+        if(!noit_hash_retrieve(&stmt_by_provider, requires, strlen(requires),
+                               &vrstmt)) {
+          noitL(noit_error,
+                "Statement %s requires %s which no one provides.\n",
+                stmt->id, requires);
+          exit(-1);
+        }
+        stmt->requires[stmt->nrequires++] = vrstmt;
+      }
+    }
+  }
+
+  /* Phase 3: Recursive sweep and mark to detect cycles.
+     We're walking the graph backwards here from dependent to provider,
+     but a cycle is a cycle, so this validates the graph. */
+  while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) {
+    stmt = vstmt;
+    if(stmt_mark_dag(stmt, ++mgen) < 0) {
+      noitL(noit_error, "Statement %s has a cyclic requirement\n", stmt->id);
+      exit(-1);
+    }
+  }
+
+  /* Phase 4: clean the markings */
+  mgen = 0;
+  memset(&iter, 0, sizeof(iter));
+  while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) {
+    stmt = vstmt;
+    stmt->marked = 0;
+  }
+
+  /* Phase 5: do the load */
+  memset(&iter, 0, sizeof(iter));
+  while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) {
+    stmt = vstmt;
+    submit_statement_node(stmt);
+  }
+
+  noit_hash_destroy(&stmt_by_provider, NULL, NULL);
+  noit_hash_destroy(&stmt_by_id, NULL, statement_node_free);
+  free(statement_configs);
-}
-
@@ -251 +443 @@
-      continue;
-    }
-
+/epl
-                             &query)) {
-      noitL(noit_iep, "No contents specified in query\n");
@@ -368 +560 @@
-     }
-#endif
+     stratcon_iep_submit_statements();
-     stratcon_datastore_iep_check_preload();
-     stratcon_iep_submit_queries();