light9 Changeset - ccd04278e357

Changeset - ccd04278e357

Parent rev.

Child rev.

[Not reviewed]

default

0 7 0

drewp@bigasterisk.com - 20 months ago 2023-06-04 00:15:40
drewp@bigasterisk.com

metrics cleanup

7 files changed with 49 insertions and 31 deletions:

light9/collector/collector.py

light9/collector/collector_client_asyncio.py

light9/collector/service.py

light9/effect/sequencer/eval_faders.py

light9/effect/sequencer/sequencer.py

light9/homepage/ServiceButtonRow.ts

light9/homepage/StatsLine.ts

0 comments (0 inline, 0 general)

light9/collector/collector.py

➞

Show inline comments

 import logging
 import time
 from typing import Dict, List, Set, Tuple, cast
 from light9.typedgraph import typedValue
 from prometheus_client import Summary
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from rdflib import URIRef
 from light9.collector.device import resolve, toOutputAttrs
 from light9.collector.output import Output as OutputInstance
 from light9.collector.weblisteners import WebListeners
 from light9.effect.settings import DeviceSettings
 from light9.namespaces import L9, RDF
 from light9.newtypes import (ClientSessionType, ClientType, DeviceAttr, DeviceClass, DeviceSetting, DeviceUri, DmxIndex, DmxMessageIndex, OutputAttr,
                              OutputRange, OutputUri, OutputValue, UnixTime, VTUnion, uriTail)
 log = logging.getLogger('collector')
 STAT_SETATTR = Summary('set_attr', 'setAttr calls')
 def makeDmxMessageIndex(base: DmxIndex, offset: DmxIndex) -> DmxMessageIndex:
     return DmxMessageIndex(base + offset - 1)
 def _outputMap(graph: SyncedGraph, outputs: Set[OutputUri]) -> Dict[Tuple[DeviceUri, OutputAttr], Tuple[OutputUri, DmxMessageIndex]]:
     """From rdf config graph, compute a map of
        (device, outputattr) : (output, index)
     that explains which output index to set for any device update.
     """
     ret = cast(Dict[Tuple[DeviceUri, OutputAttr], Tuple[OutputUri, DmxMessageIndex]], {})
@@ @@ -90,24 +92,25 @@ class Collector: @@
         ret = {}
         for output in self.outputs:
             ret[OutputUri(output.uri)] = output
         return ret
     def _compileRemapForDevice(self, dev: DeviceUri):
         for remap in self.graph.objects(dev, L9['outputAttrRange']):
             attr = typedValue(OutputAttr, self.graph, remap, L9['outputAttr'])
             start = typedValue(float, self.graph, remap, L9['start'])
             end = typedValue(float, self.graph, remap, L9['end'])
             self.remapOut[(dev, attr)] = OutputRange((start, end))
     @STAT_SETATTR.time()
     def setAttrs(self, client: ClientType, clientSession: ClientSessionType, settings: DeviceSettings, sendTime: UnixTime):
         """
         Given DeviceSettings, we resolve conflicting values,
         process them into output attrs, and call Output.update
         to send the new outputs.
         client is a string naming the type of client.
         (client, clientSession) is a unique client instance.
         clientSession is deprecated.
         Each client session's last settings will be forgotten
         after clientTimeoutSec.

light9/collector/collector_client_asyncio.py

➞

Show inline comments

 import asyncio
 import json
 import logging
 import time
 from light9 import networking
 from light9.effect.settings import DeviceSettings
 import zmq.asyncio
 from prometheus_client import Summary
 log = logging.getLogger('coll_client')
-SESS = Summary('coll_client_new_session', 'aiohttp.ClientSession')
+ZMQ_SEND = Summary('zmq_send', 'calls')
 def toCollectorJson(client, session, settings: DeviceSettings) -> str:
     assert isinstance(settings, DeviceSettings)
     return json.dumps({
         'settings': settings.asList(),
         'client': client,
         'clientSession': session,
         'sendTime': time.time(),
     })
 class _Sender:
     def __init__(self):
         self.context = zmq.asyncio.Context()
         self.socket = self.context.socket(zmq.PUB)
         self.socket.connect('tcp://127.0.0.1:9203')  #todo: tie to :collectorZmq in graph
         # old version used: 'tcp://%s:%s' % (service.host, service.port)
     @ZMQ_SEND.time()
     async def send(self, client: str, session: str, settings: DeviceSettings):
         msg = toCollectorJson(client, session, settings).encode('utf8')
         # log.info(f'zmq send {len(msg)}')
         await self.socket.send_multipart([b'setAttr', msg])
 _sender = _Sender()
 sendToCollector = _sender.send

light9/collector/service.py

➞

Show inline comments

@@ @@ -11,39 +11,37 @@ import functools @@
 import logging
 import subprocess
 import traceback
 from typing import List
 from light9 import networking
 from light9.collector.collector import Collector
 from light9.collector.output import ArtnetDmx, DummyOutput, Output, Udmx  # noqa
 from light9.collector.weblisteners import UiListener, WebListeners
 from light9.namespaces import L9
 from light9.run_local import log
 from light9.zmqtransport import parseJsonMessage
 from prometheus_client import Summary
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from starlette.applications import Starlette
 from starlette.endpoints import WebSocketEndpoint
 from starlette.requests import ClientDisconnect
 from starlette.responses import Response
 from starlette.routing import Route, WebSocketRoute
 from starlette.types import Receive, Scope, Send
 from starlette.websockets import WebSocket
 from starlette_exporter import PrometheusMiddleware, handle_metrics
 import zmq
 import zmq.asyncio
 STAT_SETATTR = Summary('set_attr', 'setAttr calls')
 # this is the rate sent to usb
 RATE = 20
 class Updates(WebSocketEndpoint, UiListener):
     def __init__(self, listeners, scope: Scope, receive: Receive, send: Send) -> None:
         super().__init__(scope, receive, send)
         self.listeners = listeners
     async def on_connect(self, websocket: WebSocket):
@@ @@ -56,33 +54,32 @@ class Updates(WebSocketEndpoint, UiListe @@
         await self.websocket.send_text(msgText)
     # async def on_receive(self, websocket, data):
     #     json.loads(data)
     async def on_disconnect(self, websocket: WebSocket, close_code: int):
         self.listeners.delClient(self)
     pass
 async def PutAttrs(collector: Collector, request):
     with STAT_SETATTR.time():
         try:
             body = await request.body()
         except ClientDisconnect:
             log.warning("PUT /attrs request disconnected- ignoring")
             return Response('', status_code=400)
         client, clientSession, settings, sendTime = parseJsonMessage(collector.graph, body)
         collector.setAttrs(client, clientSession, settings, sendTime)
         return Response('', status_code=202)
     try:
         body = await request.body()
     except ClientDisconnect:
         log.warning("PUT /attrs request disconnected- ignoring")
         return Response('', status_code=400)
     client, clientSession, settings, sendTime = parseJsonMessage(collector.graph, body)
     collector.setAttrs(client, clientSession, settings, sendTime)
     return Response('', status_code=202)
 async def zmqListener(collector):
     try:
         ctx = zmq.asyncio.Context()
         sock = ctx.socket(zmq.SUB)
         sock.bind('tcp://127.0.0.1:9203')
         sock.subscribe(b'setAttr')
         while True:
             [topic, msg] = await sock.recv_multipart()
             if topic != b'setAttr':
                 raise ValueError(topic)

light9/effect/sequencer/eval_faders.py

➞

Show inline comments

@@ @@ -9,26 +9,26 @@ from rdfdb import SyncedGraph @@
 from rdflib import URIRef
 from rdflib.term import Node
 from light9.effect.effect_function_library import EffectFunctionLibrary
 from light9.effect.effecteval2 import EffectEval2
 from light9.effect.settings import DeviceSettings, EffectSettings
 from light9.namespaces import L9, RDF
 from light9.newtypes import EffectAttr, EffectUri, UnixTime
 from light9.typedgraph import typedValue
 log = logging.getLogger('seq.fader')
 COMPILE = Summary('compile_graph_fader', '')
 COMPUTE_ALL_FADERS = Summary('compute_all_faders', '')
 COMPILE = Summary('compile_graph_fader', 'compile')
 COMPUTE_ALL_FADERS = Summary('compute_all_faders', 'compile')
 @dataclass
 class Fader:
     graph: SyncedGraph
     lib: EffectFunctionLibrary
     uri: URIRef
     effect: EffectUri
     setEffectAttr: EffectAttr
     value: Optional[float] = None  # mutable
     def __post_init__(self):

light9/effect/sequencer/sequencer.py

➞

Show inline comments

@@ @@ -74,35 +74,33 @@ class Sequencer: @@
         self.simpleOutputs = SimpleOutputs(self.graph)
         self.graph.addHandler(self.compileGraph)
         self.lastLoopSucceeded = False
         # self.codeWatcher = CodeWatcher(onChange=self.onCodeChange)
         asyncio.create_task(self.updateLoop())
     def onCodeChange(self):
         log.debug('seq.onCodeChange')
         self.graph.addHandler(self.compileGraph)
         #self.updateLoop()
     @metrics('compile_graph').time()
     def compileGraph(self) -> None:
         """rebuild our data from the graph"""
         for song in self.graph.subjects(RDF.type, L9['Song']):
             def compileSong(song: Song = cast(Song, song)) -> None:
                 self.compileSong(song)
             self.graph.addHandler(compileSong)
     @metrics('compile_song').time()
     def compileSong(self, song: Song) -> None:
         anyErrors = False
         self.notes[song] = []
         for note in self.graph.objects(song, L9['note']):
             try:
                 n = Note(self.graph, NoteUri(cast(NoteUri, note)))
             except Exception:
                 log.warn(f"failed to build Note {note} - skipping")
                 anyErrors = True
                 continue
             self.notes[song].append(n)
         if not anyErrors:
@@ @@ -122,25 +120,24 @@ class Sequencer: @@
             else:
                 took = time.time() - frameStart
                 metrics('update_loop_latency').observe(took)
                 if not self.lastLoopSucceeded:
                     log.info('Sequencer.update is working')
                     self.lastLoopSucceeded = True
                 delay = max(0, 1 / self.fps - took)
                 await asyncio.sleep(delay)
                 continue
     @metrics('update_call').time()
     async def update(self):
         with metrics('update_s0_getMusic').time():
             musicState = {'t':123.0,'song':'http://light9.bigasterisk.com/show/dance2019/song5'}#self.music.getLatest()
             if not musicState.get('song') or not isinstance(
                     musicState.get('t'), float):
                 return
             song = Song(URIRef(musicState['song']))
             # print('dispsend')
             # import pdb;pdb.set_trace()
             dispatcher.send(StateUpdate,
                             update={
                                 'song': str(song),

light9/homepage/ServiceButtonRow.ts

➞

Show inline comments

@@ @@ -15,24 +15,25 @@ export class ServiceButtonRow extends Li @@
       a {
         color: #7d7dec;
+      }
       div {
         display: flex;
         justify-content: space-between;
         padding: 2px 3px;
+      }
       .left {
         display: inline-block;
         margin-right: 3px;
         flex-grow: 1;
         min-width: 9em;
+      }
       .window {
+      }
       .serviceGrid > td {
         border: 5px solid red;
         display: inline-block;
+      }
       .big {
         font-size: 120%;
         display: inline-block;
         padding: 10px 0;
+      }

light9/homepage/StatsLine.ts

➞

Show inline comments

@@ @@ -49,46 +49,47 @@ export class StatsLine extends LitElemen @@
       if (propName == "name") {
         const reload = () => {
           fetch(this.name + "/metrics").then((resp) => {
             if (resp.ok) {
               resp
                 .text()
                 .then((msg) => {
                   this.stats = parsePrometheusTextFormat(msg) as Metrics;
                   this.extractProcessStats(this.stats);
                   setTimeout(reload, 1000);
                 })
                 .catch((err) => {
                   log(`${this.name} failing`, err)
                   setTimeout(reload, 1000);
                 });
             } else {
               if (resp.status == 502) {
                 setTimeout(reload, 5000);
+              }
               // 404: likely not mapped to a responding server
+            }
           });
         };
         reload();
+      }
     });
+  }
   extractProcessStats(stats: Metrics) {
     stats.forEach((row) => {
+    stats.forEach((row: Metric) => {
       if (row.name == "process_resident_memory_bytes") {
         this.mem = parseFloat(row.metrics[0].value) / 1024 / 1024;
+        this.mem = parseFloat(row.metrics[0].value!) / 1024 / 1024;
+      }
       if (row.name == "process_cpu_seconds_total") {
         const now = Date.now() / 1000;
         const cpuSecondsTotal = parseFloat(row.metrics[0].value);
+        const cpuSecondsTotal = parseFloat(row.metrics[0].value!);
         this.cpu = (cpuSecondsTotal - this.prevCpuTotal) / (now - this.prevCpuNow);
         this.prevCpuTotal = cpuSecondsTotal;
         this.prevCpuNow = now;
+      }
     });
+  }
   static styles = [
     css`
       :host {
         border: 2px solid #46a79f;
         display: inline-block;
@@ @@ -97,24 +98,25 @@ export class StatsLine extends LitElemen @@
         border-collapse: collapse;
         background: #000;
         color: #ccc;
         font-family: sans-serif;
+      }
       th,
       td {
         outline: 1px solid #000;
+      }
       th {
         padding: 2px 4px;
         background: #2f2f2f;
         text-align: left;
+      }
       td {
         padding: 0;
         vertical-align: top;
         text-align: center;
+      }
       td.val {
         padding: 2px 4px;
         background: #3b5651;
+      }
       .recents {
         display: flex;
@@ @@ -195,93 +197,110 @@ export class StatsLine extends LitElemen @@
   valueDisplay(m: Metric, v: Value): TemplateResult {
     if (m.type == "GAUGE") {
       return html`${v.value}`;
     } else if (m.type == "COUNTER") {
       return html`${v.value}`;
     } else if (m.type == "HISTOGRAM") {
       return this.histoDisplay(v.buckets!);
     } else if (m.type == "UNTYPED") {
       return html`${v.value}`;
     } else if (m.type == "SUMMARY") {
       log(v);
       if (!v.count) {
         return html`err: summary without count`;
+      }
-      return html`c=${v.count} percall=${((v.count && v.sum ? v.sum / v.count : 0) * 1000).toPrecision(3)}ms`;
+      return html`n=${v.count} percall=${((v.count && v.sum ? v.sum / v.count : 0) * 1000).toPrecision(3)}ms`;
     } else {
       throw m.type;
+    }
+  }
   private histoDisplay(b: { [value: string]: string }) {
     const lines: TemplateResult[] = [];
     let firstLevel;
     let lastLevel;
     let prev = 0;
     let maxDelta = 0;
     for (let level in b) {
       if (firstLevel === undefined) firstLevel = level;
       lastLevel = level;
       let count = parseFloat(b[level]);
       let delta = count - prev;
       prev = count;
       if (delta > maxDelta) maxDelta = delta;
+    }
     prev = 0;
-    const maxBarH = 60;
+    const maxBarH = 30;
     for (let level in b) {
       let count = parseFloat(b[level]);
       let delta = count - prev;
       prev = count;
       let levelf = parseFloat(level);
       const h = clamp((delta / maxDelta) * maxBarH, 1, maxBarH);
       lines.push(
         html`<div
           title="bucket=${level} count=${count}"
           style="background: yellow; margin-right: 1px; width: 8px; height: ${h}px; display: inline-block"
         ></div>`
       );
+    }
     return html`${firstLevel} ${lines} ${lastLevel}`;
+  }
   tightLabel(labs: { [key: string]: string }): string {
     const d: { [key: string]: string } = {}
     for (let k in labs) {
       if (k == 'app_name') continue;
       if (k == 'output') continue;
       if (k=='status_code'&&labs[k]=="200") continue;
       d[k] = labs[k]
+    }
     const ret = JSON.stringify(d)
     return ret == "{}" ? "" : ret
+  }
   tightMetric(name: string): string {
     return name
     .replace('starlette', '⭐')
     .replace("_request" ,"_req")
     .replace("_duration" ,"_dur")
     .replace('_seconds', '_s')
+  }
   render() {
     const now = Date.now() / 1000;
     const displayedStats = this.stats.filter(nonBoring);
     return html`
       <div>
         <table>
           ${displayedStats.map(
-            (row, rowNum) => html`
       (row, rowNum) => html`
               <tr>
-                <th>${row.type.slice(0, 1)} ${row.name}</th>
+                <th>${this.tightMetric(row.name)}</th>
                 <td>
                   <table>
                     ${row.metrics.map(
-                      (v) => html`
         (v) => html`
                         <tr>
-                          <td>${JSON.stringify(v.labels)}</td>
+                          <td>${this.tightLabel(v.labels)}</td>
                           <td>${this.valueDisplay(row, v)}</td>
                         </tr>
+                      `
-                    )}
       )}
                   </table>
                 </td>
                 ${rowNum == 0
-                  ? html`
           ? html`
                       <td rowspan="${displayedStats.length}">
                         <stats-process id="proc" cpu="${this.cpu}" mem="${this.mem}"></stats-process>
                       </td>
+                    `
-                  : ""}
           : ""}
               </tr>
+            `
-          )}
     )}
         </table>
       </div>
     `;
+  }
+}

0 comments (0 inline, 0 general)