#94 indicate outdated cluster data by pulsating

app/src/app.js

@@ -25,7 +25,10 @@ export default class App {
         this.keepAliveSeconds = 20
         // always reconnect after 5 minutes
         this.maxConnectionLifetimeSeconds = 300
+        // consider cluster data older than 1 minute outdated
+        this.maxDataAgeSeconds = 60
         this.clusters = new Map()
+        this.clusterStatuses = new Map()
     }
 
     parseLocationHash() {
@@ -122,6 +125,8 @@ export default class App {
         addEventListener(
             'keydown', downHandler.bind(this), false
         )
+
+        setInterval(this.pruneUnavailableClusters.bind(this), 5 * 1000)
     }
 
     draw() {
@@ -313,12 +318,14 @@ export default class App {
         for (const cluster of clusters) {
             if (!this.selectedClusters.size || this.selectedClusters.has(cluster.id)) {
                 clusterIds.add(cluster.id)
+                const status = this.clusterStatuses.get(cluster.id)
                 let clusterBox = clusterComponentById[cluster.id]
                 if (!clusterBox) {
-                    clusterBox = new Cluster(cluster, this.tooltip)
+                    clusterBox = new Cluster(cluster, status, this.tooltip)
                     this.viewContainer.addChild(clusterBox)
                 } else {
                     clusterBox.cluster = cluster
+                    clusterBox.status = status
                 }
                 clusterBox.draw()
                 clusterBox.x = 0
@@ -391,6 +398,23 @@ export default class App {
         }
     }
 
+    pruneUnavailableClusters() {
+        let updateNeeded = false
+        const nowSeconds = Date.now() / 1000
+        for (const [clusterId, statusObj] of this.clusterStatuses.entries()) {
+            const lastQueryTime = statusObj.last_query_time || 0
+            if (lastQueryTime < nowSeconds - this.maxDataAgeSeconds) {
+                this.clusters.delete(clusterId)
+                updateNeeded = true
+            } else if (lastQueryTime < nowSeconds - 20) {
+                updateNeeded = true
+            }
+        }
+        if (updateNeeded) {
+            this.update()
+        }
+    }
+
     disconnect() {
         if (this.eventSource != null) {
             this.eventSource.close()
@@ -399,6 +423,15 @@ export default class App {
         }
     }
 
+    refreshLastQueryTime(clusterId) {
+        let statusObj = this.clusterStatuses.get(clusterId)
+        if (!statusObj) {
+            statusObj = {}
+        }
+        statusObj.last_query_time = Date.now() / 1000
+        this.clusterStatuses.set(clusterId, statusObj)
+    }
+
     connect() {
         // first close the old connection
         this.disconnect()
@@ -425,13 +458,21 @@ export default class App {
             that._errors = 0
             that.keepAlive()
             const cluster = JSON.parse(event.data)
+            const status = that.clusterStatuses.get(cluster.id)
+            const nowSeconds = Date.now() / 1000
+            if (status && status.last_query_time < nowSeconds - that.maxDataAgeSeconds) {
+                // outdated data => ignore
+            } else {
                 that.clusters.set(cluster.id, cluster)
                 that.update()
+            }
         })
         eventSource.addEventListener('clusterdelta', function(event) {
             that._errors = 0
             that.keepAlive()
             const data = JSON.parse(event.data)
+            // we received some delta => we know that the cluster query succeeded!
+            that.refreshLastQueryTime(data.cluster_id)
             let cluster = that.clusters.get(data.cluster_id)
             if (cluster && data.delta) {
                 // deep copy cluster object (patch function mutates inplace!)
@@ -441,6 +482,12 @@ export default class App {
                 that.update()
             }
         })
+        eventSource.addEventListener('clusterstatus', function(event) {
+            that._errors = 0
+            that.keepAlive()
+            const data = JSON.parse(event.data)
+            that.clusterStatuses.set(data.cluster_id, data.status)
+        })
         this.connectTime = Date.now()
     }
 

app/src/cluster.js

@@ -4,12 +4,25 @@ import App from './app.js'
 const PIXI = require('pixi.js')
 
 export default class Cluster extends PIXI.Graphics {
-    constructor (cluster, tooltip) {
+    constructor (cluster, status, tooltip) {
         super()
         this.cluster = cluster
+        this.status = status
         this.tooltip = tooltip
     }
 
+    destroy() {
+        if (this.tick) {
+            PIXI.ticker.shared.remove(this.tick, this)
+        }
+        super.destroy()
+    }
+
+    pulsate(_time) {
+        const v = Math.sin((PIXI.ticker.shared.lastTime % 1000) / 1000. * Math.PI)
+        this.alpha = 0.4 + (v * 0.6)
+    }
+
     draw () {
         this.removeChildren()
         this.clear()
@@ -80,7 +93,7 @@ export default class Cluster extends PIXI.Graphics {
         const width = Math.max(masterWidth, workerWidth)
         this.drawRect(0, 0, width, top + masterHeight + workerHeight)
 
-        var topHandle = new PIXI.Graphics()
+        const topHandle = this.topHandle = new PIXI.Graphics()
         topHandle.beginFill(App.current.theme.primaryColor, 1)
         topHandle.drawRect(0, 0, width, 15)
         topHandle.endFill()
@@ -90,11 +103,29 @@ export default class Cluster extends PIXI.Graphics {
         topHandle.on('click', function(_event) {
             App.current.toggleCluster(that.cluster.id)
         })
-        var text = new PIXI.Text(this.cluster.api_server_url, {fontFamily: 'ShareTechMono', fontSize: 10, fill: 0x000000})
+        const text = new PIXI.Text(this.cluster.api_server_url, {fontFamily: 'ShareTechMono', fontSize: 10, fill: 0x000000})
         text.x = 2
         text.y = 2
         topHandle.addChild(text)
         this.addChild(topHandle)
+
+        let newTick = null
+        const nowSeconds = Date.now() / 1000
+        if (this.status && this.status.last_query_time < nowSeconds - 20) {
+            newTick = this.pulsate
+        }
+
+        if (newTick && newTick != this.tick) {
+            this.tick = newTick
+            // important: only register new listener if it does not exist yet!
+            // (otherwise we leak listeners)
+            PIXI.ticker.shared.add(this.tick, this)
+        } else if (!newTick && this.tick) {
+            PIXI.ticker.shared.remove(this.tick, this)
+            this.tick = null
+            this.alpha = 1
+            this.tint = 0xffffff
+        }
     }
 
 }

kube_ops_view/kubernetes.py

@@ -110,4 +110,9 @@ def query_kubernetes_cluster(cluster):
                                 container['resources']['usage'] = container_metrics['usage']
     except Exception as e:
         logger.warning('Failed to query pod metrics for cluster {}: {}'.format(cluster.id, get_short_error_message(e)))
-    return {'id': cluster_id, 'api_server_url': api_server_url, 'nodes': nodes, 'unassigned_pods': unassigned_pods}
+    return {
+        'id': cluster_id,
+        'api_server_url': api_server_url,
+        'nodes': nodes,
+        'unassigned_pods': unassigned_pods
+    }

kube_ops_view/main.py

@@ -88,9 +88,15 @@ def event(cluster_ids: set):
     # first sent full data once
     for cluster_id in app.store.get_cluster_ids():
         if not cluster_ids or cluster_id in cluster_ids:
+            status = app.store.get_cluster_status(cluster_id)
+            if status:
+                # send the cluster status including last_query_time BEFORE the cluster data
+                # so the UI knows how to render correctly from the start
+                yield 'event: clusterstatus\ndata: ' + json.dumps({'cluster_id': cluster_id, 'status': status}, separators=(',', ':')) + '\n\n'
             cluster = app.store.get_cluster_data(cluster_id)
             if cluster:
                 yield 'event: clusterupdate\ndata: ' + json.dumps(cluster, separators=(',', ':')) + '\n\n'
+
     while True:
         for event_type, event_data in app.store.listen():
             # hacky, event_data can be delta or full cluster object

kube_ops_view/update.py

@@ -55,6 +55,7 @@ def update_clusters(cluster_discoverer, query_cluster: callable, store, query_in
                     except Exception as e:
                         backoff = handle_query_failure(e, cluster, backoff)
                         status['backoff'] = backoff
+                        store.publish('clusterstatus', {'cluster_id': cluster.id, 'status': status})
                     else:
                         status['last_query_time'] = now
                         if backoff:
@@ -70,6 +71,8 @@ def update_clusters(cluster_discoverer, query_cluster: callable, store, query_in
                                 store.set_cluster_data(cluster.id, data)
                         else:
                             logger.info('Discovered new cluster {} ({}).'.format(cluster.id, cluster.api_server_url))
+                            # first send status with last_query_time!
+                            store.publish('clusterstatus', {'cluster_id': cluster.id, 'status': status})
                             store.publish('clusterupdate', data)
                             store.set_cluster_data(cluster.id, data)
                     store.set_cluster_status(cluster.id, status)