Mercurial > code > home > repos > homeauto

annotate service/collector/sse_collector.py @ 1245:4c123099e5b6

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
collector build improvements; stats and logging Ignore-this: 22fbbb2cde6a6bdbd9d0017f1f157a41 darcs-hash:3bcbb4fa359c1165e1c900ce0264869567e1d1fa
author drewp <drewp@bigasterisk.com>
date Thu, 18 Apr 2019 09:15:39 -0700
parents 24c004aac998
children f18b95f81d43
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
rev   line source
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
1 from __future__ import division
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
2 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
3 requesting /graph/foo returns an SSE patch stream that's the
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
4 result of fetching multiple other SSE patch streams. The result stream
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
5 may include new statements injected by this service.
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
6
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
7 Future:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
8 - filter out unneeded stmts from the sources
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
9 - give a time resolution and concatenate any patches that come faster than that res
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
10 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
11 from crochet import no_setup
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
12 no_setup()
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
13
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
14 import sys, logging, collections, json, time
1245
4c123099e5b6 collector build improvements; stats and logging
drewp <drewp@bigasterisk.com>
parents: 1242
diff changeset
15 from twisted.internet import reactor, defer
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
16 import cyclone.web, cyclone.sse
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
17 from rdflib import URIRef, Namespace
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
18 from docopt import docopt
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
19
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
20 sys.path.append('/opt') # docker is putting ../../lib/ here
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
21 from logsetup import log
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
22 from patchablegraph import jsonFromPatch
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
23
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
24 from rdfdb.patch import Patch
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
25
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
26 from patchsource import ReconnectingPatchSource
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
27
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
28 ROOM = Namespace("http://projects.bigasterisk.com/room/")
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
29 COLLECTOR = URIRef('http://bigasterisk.com/sse_collector/')
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
30
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
31 from sse_collector_config import config
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
32
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
33 class LocalStatements(object):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
34 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
35 functions that make statements originating from sse_collector itself
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
36 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
37 def __init__(self, applyPatch):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
38 self.applyPatch = applyPatch
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
39 self._sourceState = {} # source: state URIRef
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
40
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
41 def setSourceState(self, source, state):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
42 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
43 add a patch to the COLLECTOR graph about the state of this
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
44 source. state=None to remove the source.
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
45 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
46 oldState = self._sourceState.get(source, None)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
47 if state == oldState:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
48 return
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
49 log.info('source state %s -> %s', source, state)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
50 if oldState is None:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
51 self._sourceState[source] = state
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
52 self.applyPatch(COLLECTOR, Patch(addQuads=[
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
53 (COLLECTOR, ROOM['source'], source, COLLECTOR),
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
54 (source, ROOM['state'], state, COLLECTOR),
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
55 ]))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
56 elif state is None:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
57 del self._sourceState[source]
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
58 self.applyPatch(COLLECTOR, Patch(delQuads=[
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
59 (COLLECTOR, ROOM['source'], source, COLLECTOR),
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
60 (source, ROOM['state'], oldState, COLLECTOR),
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
61 ]))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
62 else:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
63 self._sourceState[source] = state
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
64 self.applyPatch(COLLECTOR, Patch(
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
65 addQuads=[
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
66 (source, ROOM['state'], state, COLLECTOR),
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
67 ],
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
68 delQuads=[
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
69 (source, ROOM['state'], oldState, COLLECTOR),
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
70 ]))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
71
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
72 def abbrevTerm(t):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
73 if isinstance(t, URIRef):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
74 return (t.replace('http://projects.bigasterisk.com/room/', 'room:')
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
75 .replace('http://bigasterisk.com/sse_collector/', 'sc:'))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
76 return t
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
77
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
78 def abbrevStmt(stmt):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
79 return '(%s %s %s %s)' % tuple(map(abbrevTerm, stmt))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
80
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
81 class ActiveStatements(object):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
82 def __init__(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
83
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
84 # This table holds statements asserted by any of our sources
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
85 # plus local statements that we introduce (source is
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
86 # http://bigasterisk.com/sse_collector/).
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
87 self.statements = collections.defaultdict(lambda: (set(), set())) # (s,p,o,c): (sourceUrls, handlers)`
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
88
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
89 def state(self):
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
90 return {
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
91 'len': len(self.statements),
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
92 }
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
93
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
94 def _postDeleteStatements(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
95 statements = self.statements
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
96 class PostDeleter(object):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
97 def __enter__(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
98 self._garbage = []
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
99 return self
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
100 def add(self, stmt):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
101 self._garbage.append(stmt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
102 def __exit__(self, type, value, traceback):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
103 if type is not None:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
104 raise
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
105 for stmt in self._garbage:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
106 del statements[stmt]
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
107 return PostDeleter()
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
108
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
109 def pprintTable(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
110 for i, (stmt, (sources, handlers)) in enumerate(sorted(self.statements.items())):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
111 print "%03d. %-80s from %s to %s" % (
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
112 i, abbrevStmt(stmt), [abbrevTerm(s) for s in sources], handlers)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
113
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
114 def makeSyncPatch(self, handler, sources):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
115 # todo: this could run all handlers at once, which is how we use it anyway
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
116 adds = []
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
117 dels = []
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
118
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
119 with self._postDeleteStatements() as garbage:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
120 for stmt, (stmtSources, handlers) in self.statements.iteritems():
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
121 belongsInHandler = not set(sources).isdisjoint(stmtSources)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
122 handlerHasIt = handler in handlers
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
123 #log.debug("%s %s %s", abbrevStmt(stmt), belongsInHandler, handlerHasIt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
124 if belongsInHandler and not handlerHasIt:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
125 adds.append(stmt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
126 handlers.add(handler)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
127 elif not belongsInHandler and handlerHasIt:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
128 dels.append(stmt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
129 handlers.remove(handler)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
130 if not handlers and not stmtSources:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
131 garbage.add(stmt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
132
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
133 return Patch(addQuads=adds, delQuads=dels)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
134
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
135 def applySourcePatch(self, source, p):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
136 for stmt in p.addQuads:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
137 sourceUrls, handlers = self.statements[stmt]
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
138 if source in sourceUrls:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
139 raise ValueError("%s added stmt that it already had: %s" %
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
140 (source, abbrevStmt(stmt)))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
141 sourceUrls.add(source)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
142
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
143 with self._postDeleteStatements() as garbage:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
144 for stmt in p.delQuads:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
145 sourceUrls, handlers = self.statements[stmt]
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
146 if source not in sourceUrls:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
147 raise ValueError("%s deleting stmt that it didn't have: %s" %
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
148 (source, abbrevStmt(stmt)))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
149 sourceUrls.remove(source)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
150 # this is rare, since some handler probably still has
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
151 # the stmt we're deleting, but it can happen e.g. when
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
152 # a handler was just deleted
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
153 if not sourceUrls and not handlers:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
154 garbage.add(stmt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
155
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
156 def replaceSourceStatements(self, source, stmts):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
157 log.debug('replaceSourceStatements with %s stmts', len(stmts))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
158 newStmts = set(stmts)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
159
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
160 with self._postDeleteStatements() as garbage:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
161 for stmt, (sources, handlers) in self.statements.iteritems():
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
162 if source in sources:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
163 if stmt not in stmts:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
164 sources.remove(source)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
165 if not sources and not handlers:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
166 garbage.add(stmt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
167 else:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
168 if stmt in stmts:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
169 sources.add(source)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
170 newStmts.discard(stmt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
171
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
172 self.applySourcePatch(source, Patch(addQuads=newStmts, delQuads=[]))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
173
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
174 def discardHandler(self, handler):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
175 with self._postDeleteStatements() as garbage:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
176 for stmt, (sources, handlers) in self.statements.iteritems():
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
177 handlers.discard(handler)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
178 if not sources and not handlers:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
179 garbage.add(stmt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
180
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
181 def discardSource(self, source):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
182 with self._postDeleteStatements() as garbage:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
183 for stmt, (sources, handlers) in self.statements.iteritems():
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
184 sources.discard(source)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
185 if not sources and not handlers:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
186 garbage.add(stmt)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
187
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
188 class GraphClients(object):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
189 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
190 All the active PatchSources and SSEHandlers
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
191
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
192 To handle all the overlapping-statement cases, we store a set of
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
193 true statements along with the sources that are currently
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
194 asserting them and the requesters who currently know them. As
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
195 statements come and go, we make patches to send to requesters.
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
196 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
197 def __init__(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
198 self.clients = {} # url: PatchSource (COLLECTOR is not listed)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
199 self.handlers = set() # handler
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
200 self.statements = ActiveStatements()
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
201
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
202 self._localStatements = LocalStatements(self._onPatch)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
203
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
204 def state(self):
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
205 return {
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
206 'clients': [ps.state() for ps in self.clients.values()],
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
207 'sseHandlers': [h.state() for h in self.handlers],
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
208 'statements': self.statements.state(),
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
209 }
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
210
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
211 def _sourcesForHandler(self, handler):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
212 streamId = handler.streamId
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
213 matches = [s for s in config['streams'] if s['id'] == streamId]
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
214 if len(matches) != 1:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
215 raise ValueError("%s matches for %r" % (len(matches), streamId))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
216 return map(URIRef, matches[0]['sources']) + [COLLECTOR]
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
217
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
218 def _onPatch(self, source, p, fullGraph=False):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
219 if fullGraph:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
220 # a reconnect may need to resend the full graph even
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
221 # though we've already sent some statements
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
222 self.statements.replaceSourceStatements(source, p.addQuads)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
223 else:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
224 self.statements.applySourcePatch(source, p)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
225
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
226 self._sendUpdatePatch()
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
227
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
228 if log.isEnabledFor(logging.DEBUG):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
229 self.statements.pprintTable()
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
230
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
231 if source != COLLECTOR:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
232 self._localStatements.setSourceState(
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
233 source,
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
234 ROOM['fullGraphReceived'] if fullGraph else
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
235 ROOM['patchesReceived'])
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
236
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
237 def _sendUpdatePatch(self, handler=None):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
238 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
239 send a patch event out this handler to bring it up to date with
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
240 self.statements
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
241 """
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
242 # reduce loops here- prepare all patches at once
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
243 for h in (self.handlers if handler is None else [handler]):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
244 p = self.statements.makeSyncPatch(h, self._sourcesForHandler(h))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
245 if not p.isNoop():
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
246 log.debug("send patch %s to %s", p.shortSummary(), h)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
247 # This can be a giant line, which was a problem once. Might be
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
248 # nice for this service to try to break it up into multiple sends,
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
249 # although there's no guarantee at all since any single stmt
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
250 # could be any length.
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
251 h.sendEvent(message=jsonFromPatch(p), event='patch')
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
252
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
253 def addSseHandler(self, handler):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
254 log.info('addSseHandler %r %r', handler, handler.streamId)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
255
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
256 # fail early if id doesn't match
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
257 sources = self._sourcesForHandler(handler)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
258
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
259 self.handlers.add(handler)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
260
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
261 for source in sources:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
262 if source not in self.clients and source != COLLECTOR:
1245
4c123099e5b6 collector build improvements; stats and logging
drewp <drewp@bigasterisk.com>
parents: 1242
diff changeset
263 log.debug('connect to patch source %s', source)
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
264 self._localStatements.setSourceState(source, ROOM['connect'])
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
265 self.clients[source] = ReconnectingPatchSource(
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
266 source, listener=lambda p, fullGraph, source=source: self._onPatch(
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
267 source, p, fullGraph))
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
268 self._sendUpdatePatch(handler)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
269
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
270 def removeSseHandler(self, handler):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
271 log.info('removeSseHandler %r', handler)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
272 self.statements.discardHandler(handler)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
273 for source in self._sourcesForHandler(handler):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
274 for otherHandler in self.handlers:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
275 if (otherHandler != handler and
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
276 source in self._sourcesForHandler(otherHandler)):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
277 break
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
278 else:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
279 self._stopClient(source)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
280
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
281 self.handlers.remove(handler)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
282
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
283 def _stopClient(self, url):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
284 if url == COLLECTOR:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
285 return
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
286
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
287 self.clients[url].stop()
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
288
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
289 self.statements.discardSource(url)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
290
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
291 self._localStatements.setSourceState(url, None)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
292 del self.clients[url]
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
293
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
294
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
295 class SomeGraph(cyclone.sse.SSEHandler):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
296 _handlerSerial = 0
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
297 def __init__(self, application, request):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
298 cyclone.sse.SSEHandler.__init__(self, application, request)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
299 self.streamId = request.uri[len('/graph/'):]
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
300 self.graphClients = self.settings.graphClients
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
301 self.created = time.time()
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
302
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
303 self._serial = SomeGraph._handlerSerial
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
304 SomeGraph._handlerSerial += 1
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
305
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
306 def __repr__(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
307 return '<Handler #%s>' % self._serial
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
308
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
309 def state(self):
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
310 return {
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
311 'created': round(self.created, 2),
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
312 'ageHours': round((time.time() - self.created) / 3600, 2),
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
313 'streamId': self.streamId,
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
314 'remoteIp': self.request.remote_ip,
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
315 'userAgent': self.request.headers.get('user-agent'),
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
316 }
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
317
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
318 def bind(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
319 self.graphClients.addSseHandler(self)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
320
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
321 def unbind(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
322 self.graphClients.removeSseHandler(self)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
323
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
324 class State(cyclone.web.RequestHandler):
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
325 @STATS.getState.time()
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
326 def get(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
327 try:
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
328 state = self.settings.graphClients.state()
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
329 except:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
330 import traceback; traceback.print_exc()
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
331 raise
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
332
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
333 self.write(json.dumps({'graphClients': state}, indent=2))
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
334
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
335 class Root(cyclone.web.RequestHandler):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
336 def get(self):
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
337 self.write('<html><body>sse_collector</body></html>')
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
338
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
339 if __name__ == '__main__':
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
340 arg = docopt("""
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
341 Usage: sse_collector.py [options]
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
342
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
343 -v Verbose
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
344 """)
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
345
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
346 if arg['-v']:
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
347 import twisted.python.log
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
348 twisted.python.log.startLogging(sys.stdout)
1245
4c123099e5b6 collector build improvements; stats and logging
drewp <drewp@bigasterisk.com>
parents: 1242
diff changeset
349 log.setLevel(logging.DEBUG)
4c123099e5b6 collector build improvements; stats and logging
drewp <drewp@bigasterisk.com>
parents: 1242
diff changeset
350 defer.setDebugging(True)
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
351
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
352
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
353 graphClients = GraphClients()
1245
4c123099e5b6 collector build improvements; stats and logging
drewp <drewp@bigasterisk.com>
parents: 1242
diff changeset
354 #exporter = InfluxExporter(... to export some stats values
4c123099e5b6 collector build improvements; stats and logging
drewp <drewp@bigasterisk.com>
parents: 1242
diff changeset
355
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
356 reactor.listenTCP(
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
357 9072,
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
358 cyclone.web.Application(
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
359 handlers=[
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
360 (r'/', Root),
1242
24c004aac998 stats->state to make room for greplin stats
drewp <drewp@bigasterisk.com>
parents: 1158
diff changeset
361 (r'/state', State),
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
362 (r'/graph/(.*)', SomeGraph),
1245
4c123099e5b6 collector build improvements; stats and logging
drewp <drewp@bigasterisk.com>
parents: 1242
diff changeset
363 (r'/stats/(.*)', StatsHandler, {'serverName': 'collector'}),
1156
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
364 ],
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
365 graphClients=graphClients),
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
366 interface='::')
ee168d55524a reasoning & collector move into docker images
drewp <drewp@bigasterisk.com>
parents:
diff changeset
367 reactor.run()