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