Improved Plotly/Dash example. It is now somehow informative.

 * TCP timeout after FIN/RST: switched back to the value from a35fc1d5ea8570609cc0c8cf6edadc81f8f5bb76
 * py-flow-info: reset 'guessed' flag after detection/detection-update received

Signed-off-by: Toni Uhlig <matzeton@googlemail.com>

3 files changed, 386 insertions, 128 deletions

diff --git a/examples/py-flow-dashboard/flow-dash.py b/examples/py-flow-dashboard/flow-dash.py
index 8e49ed020..283a97bb9 100755
--- a/examples/py-flow-dashboard/flow-dash.py
+++ b/examples/py-flow-dashboard/flow-dash.py
@@ -1,16 +1,11 @@
 #!/usr/bin/env python3
 
-from collections import deque
-import dash
-from dash.dependencies import Output, Input
-import dash_core_components as dcc
-import dash_html_components as html
 import multiprocessing
 import os
-import plotly
-import plotly.graph_objs as go
 import sys
 
+import plotly_dash
+
 sys.path.append(os.path.dirname(sys.argv[0]) + '/../share/nDPId')
 sys.path.append(os.path.dirname(sys.argv[0]) + '/../usr/share/nDPId')
 try:
@@ -21,146 +16,109 @@ except ImportError:
     import nDPIsrvd
     from nDPIsrvd import nDPIsrvdSocket
 
-mgr = multiprocessing.Manager()
-
-global shared_flow_dict
-shared_flow_dict = mgr.dict()
-
-FLOW_COUNT_DATAPOINTS = 50
-
-global live_flow_count_X
-live_flow_count_X = deque(maxlen=FLOW_COUNT_DATAPOINTS)
-live_flow_count_X.append(1)
-global live_flow_count_Y
-live_flow_count_Y = deque(maxlen=FLOW_COUNT_DATAPOINTS)
-live_flow_count_Y.append(1)
-
-live_flow_bars = ['risky', 'midstream', 'detected', 'guessed', 'not-detected']
-fig = go.Figure()
-
-app = dash.Dash(__name__)
-app.layout = html.Div(
-    [
-        dcc.Graph(id='live-flow-count', animate=True),
-        dcc.Graph(id='live-flow-bars',  animate=True, figure=fig),
-        dcc.Interval(
-            id='graph-update',
-            interval=1000,
-            n_intervals=0
-        ),
-    ]
-)
-
-
-@app.callback(
-    Output('live-flow-count', 'figure'),
-    [Input('graph-update', 'n_intervals')]
-)
-def update_graph_scatter(n):
-    live_flow_count_X.append(live_flow_count_X[-1]+1)
-    live_flow_count_Y.append(len(shared_flow_dict))
-
-    data = plotly.graph_objs.Scatter(
-        x=list(live_flow_count_X),
-        y=list(live_flow_count_Y),
-        name='Scatter',
-        mode='lines+markers'
-    )
-
-    return {
-            'data': [data],
-            'layout':
-            go.Layout(
-                xaxis=dict(
-                    range=[min(live_flow_count_X), max(live_flow_count_X)]
-                ),
-                yaxis=dict(
-                    range=[min(live_flow_count_Y), max(live_flow_count_Y)]
-                ),
-            )}
-
-
-@app.callback(
-    Output('live-flow-bars', 'figure'),
-    [Input('graph-update', 'n_intervals')]
-)
-def update_pie(n):
-    values = [0, 0, 0, 0, 0]
-
-    for flow_id in shared_flow_dict.keys():
-        try:
-            flow = shared_flow_dict[flow_id]
-        except KeyError:
-            continue
-
-        if flow['is_risky'] is True:
-            values[0] += 1
-
-        if flow['is_midstream'] is True:
-            values[1] += 1
-
-        if flow['is_detected'] is True:
-            values[2] += 1
-
-        if flow['is_guessed'] is True:
-            values[3] += 1
-
-        if flow['is_not_detected'] is True:
-            values[4] += 1
-
-    # print(values)
-
-    return {
-            'data': [
-                        go.Bar(name='', x=live_flow_bars, y=values)
-                    ],
-            'layout': go.Layout(yaxis=dict(range=[0, max(values)]))
-           }
-
-
-def web_worker():
-    app.run_server()
-
 
 def nDPIsrvd_worker_onFlowCleanup(instance, current_flow, global_user_data):
+    _, shared_flow_dict = global_user_data
+
+    flow_id = current_flow.flow_id
+
+    shared_flow_dict['current-flows'] -= 1
+
+    if shared_flow_dict[flow_id]['is_detected'] is True:
+        shared_flow_dict['current-detected-flows'] -= 1
+
+    if shared_flow_dict[flow_id]['is_guessed'] is True:
+        shared_flow_dict['current-guessed-flows'] -= 1
+
+    if shared_flow_dict[flow_id]['is_not_detected'] is True:
+        shared_flow_dict['current-detected-flows'] -= 1
+
+    if shared_flow_dict[flow_id]['is_midstream'] is True:
+        shared_flow_dict['current-midstream-flows'] -= 1
+
+    if shared_flow_dict[flow_id]['is_risky'] is True:
+        shared_flow_dict['current-risky-flows'] -= 1
+
     del shared_flow_dict[current_flow.flow_id]
 
     return True
 
 def nDPIsrvd_worker_onJsonLineRecvd(json_dict, instance, current_flow, global_user_data):
+    nsock, shared_flow_dict = global_user_data
+
+    shared_flow_dict['total-events'] += 1
+    shared_flow_dict['total-bytes']   = nsock.received_bytes
+
     if 'flow_id' not in json_dict:
         return True
+    else:
+        if current_flow.flow_id != json_dict['flow_id']:
+            return False
+        flow_id = current_flow.flow_id
 
     # print(json_dict)
 
-    if json_dict['flow_id'] not in shared_flow_dict:
-        shared_flow_dict[json_dict['flow_id']] = mgr.dict()
-        shared_flow_dict[json_dict['flow_id']]['is_detected'] = False
-        shared_flow_dict[json_dict['flow_id']]['is_guessed'] = False
-        shared_flow_dict[json_dict['flow_id']]['is_not_detected'] = False
-        shared_flow_dict[json_dict['flow_id']]['is_midstream'] = False
-        shared_flow_dict[json_dict['flow_id']]['is_risky'] = False
+    if flow_id not in shared_flow_dict:
+        shared_flow_dict[flow_id] = mgr.dict()
+        shared_flow_dict[flow_id]['is_detected']     = False
+        shared_flow_dict[flow_id]['is_guessed']      = False
+        shared_flow_dict[flow_id]['is_not_detected'] = False
+        shared_flow_dict[flow_id]['is_midstream']    = False
+        shared_flow_dict[flow_id]['is_risky']        = False
+
+        shared_flow_dict['total-flows']   += 1
+        shared_flow_dict['current-flows'] += 1
+
+    if 'midstream' in json_dict and json_dict['midstream'] != 0:
+        if shared_flow_dict[flow_id]['is_midstream'] is False:
+            shared_flow_dict['total-midstream-flows']   += 1
+            shared_flow_dict['current-midstream-flows'] += 1
+        shared_flow_dict[flow_id]['is_midstream'] = True
+
+    if 'ndpi' in json_dict and 'flow_risk' in json_dict['ndpi']:
+        if shared_flow_dict[flow_id]['is_risky'] is False:
+            shared_flow_dict['total-risky-flows']   += 1
+            shared_flow_dict['current-risky-flows'] += 1
+        shared_flow_dict[flow_id]['is_risky'] = True
 
     if 'flow_event_name' not in json_dict:
         return True
 
     if json_dict['flow_event_name'] == 'new':
-        if 'midstream' in json_dict and json_dict['midstream'] != 0:
-            shared_flow_dict[json_dict['flow_id']]['is_midstream'] = True
+
+        pass
+
     elif json_dict['flow_event_name'] == 'guessed':
-        shared_flow_dict[json_dict['flow_id']]['is_guessed'] = True
+
+        if shared_flow_dict[flow_id]['is_guessed'] is False:
+            shared_flow_dict['total-guessed-flows']   += 1
+            shared_flow_dict['current-guessed-flows'] += 1
+        shared_flow_dict[flow_id]['is_guessed'] = True
+
     elif json_dict['flow_event_name'] == 'not-detected':
-        shared_flow_dict[json_dict['flow_id']]['is_not_detected'] = True
-    elif json_dict['flow_event_name'] == 'detected':
-        shared_flow_dict[json_dict['flow_id']]['is_detected'] = True
-        shared_flow_dict[json_dict['flow_id']]['is_guessed'] = False
-        if 'ndpi' in json_dict and 'flow_risk' in json_dict['ndpi']:
-            shared_flow_dict[json_dict['flow_id']]['is_risky'] = True
+
+        if shared_flow_dict[flow_id]['is_not_detected'] is False:
+            shared_flow_dict['total-not-detected-flows']   += 1
+            shared_flow_dict['current-not-detected-flows'] += 1
+        shared_flow_dict[flow_id]['is_not_detected'] = True
+
+    elif json_dict['flow_event_name'] == 'detected' or \
+         json_dict['flow_event_name'] == 'detection-update':
+
+        if shared_flow_dict[flow_id]['is_detected'] is False:
+            shared_flow_dict['total-detected-flows']   += 1
+            shared_flow_dict['current-detected-flows'] += 1
+        shared_flow_dict[flow_id]['is_detected'] = True
+
+        if shared_flow_dict[flow_id]['is_guessed'] is True:
+            shared_flow_dict['total-guessed-flows']   -= 1
+            shared_flow_dict['current-guessed-flows'] -= 1
+        shared_flow_dict[flow_id]['is_guessed'] = False
 
     return True
 
 
-def nDPIsrvd_worker(address, nDPIsrvd_global_user_data):
+def nDPIsrvd_worker(address, shared_flow_dict):
     sys.stderr.write('Recv buffer size: {}\n'
                      .format(nDPIsrvd.NETWORK_BUFFER_MAX_SIZE))
     sys.stderr.write('Connecting to {} ..\n'
@@ -171,7 +129,7 @@ def nDPIsrvd_worker(address, nDPIsrvd_global_user_data):
     nsock.connect(address)
     nsock.loop(nDPIsrvd_worker_onJsonLineRecvd,
                nDPIsrvd_worker_onFlowCleanup,
-               nDPIsrvd_global_user_data)
+               (nsock, shared_flow_dict))
 
 
 if __name__ == '__main__':
@@ -179,11 +137,31 @@ if __name__ == '__main__':
     args = argparser.parse_args()
     address = nDPIsrvd.validateAddress(args)
 
+    mgr = multiprocessing.Manager()
+    shared_flow_dict = mgr.dict()
+
+    shared_flow_dict['total-events']             = 0
+    shared_flow_dict['total-bytes']              = 0
+    shared_flow_dict['total-flows']              = 0
+    shared_flow_dict['total-detected-flows']     = 0
+    shared_flow_dict['total-risky-flows']        = 0
+    shared_flow_dict['total-midstream-flows']    = 0
+    shared_flow_dict['total-guessed-flows']      = 0
+    shared_flow_dict['total-not-detected-flows'] = 0
+
+    shared_flow_dict['current-flows']              = 0
+    shared_flow_dict['current-detected-flows']     = 0
+    shared_flow_dict['current-risky-flows']        = 0
+    shared_flow_dict['current-midstream-flows']    = 0
+    shared_flow_dict['current-guessed-flows']      = 0
+    shared_flow_dict['current-not-detected-flows'] = 0
+
     nDPIsrvd_job = multiprocessing.Process(target=nDPIsrvd_worker,
-                                           args=(address, None))
+                                           args=(address, shared_flow_dict))
     nDPIsrvd_job.start()
 
-    web_job = multiprocessing.Process(target=web_worker, args=())
+    web_job = multiprocessing.Process(target=plotly_dash.web_worker,
+                                      args=(shared_flow_dict,))
     web_job.start()
 
     nDPIsrvd_job.join()
diff --git a/examples/py-flow-dashboard/plotly_dash.py b/examples/py-flow-dashboard/plotly_dash.py
new file mode 100644
index 000000000..c4cf59086
--- /dev/null
+++ b/examples/py-flow-dashboard/plotly_dash.py
@@ -0,0 +1,276 @@
+import math
+
+import dash
+from dash.dependencies import Input, Output, State
+import dash_core_components as dcc
+import dash_html_components as html
+import dash_daq as daq
+import dash_table as dt
+
+import plotly.graph_objects as go
+
+global shared_flow_dict
+
+app = dash.Dash(__name__)
+
+def generate_box():
+    return { \
+        'display': 'flex', 'flex-direction': 'row', \
+        'box-shadow': '0 1px 3px rgba(0,0,0,0.12), 0 1px 2px rgba(0,0,0,0.24)', \
+        'background-color': '#082255' \
+    }
+
+def generate_led_display(div_id, label_name):
+    return daq.LEDDisplay( \
+        id=div_id, \
+        label={'label': label_name, 'style': {'color': '#C4CDD5'}}, \
+        labelPosition='bottom', \
+        value='0', \
+        backgroundColor='#082255', \
+        color='#C4CDD5', \
+    )
+
+def generate_gauge(div_id, label_name, max_value=10):
+    return daq.Gauge( \
+        id=div_id, \
+        value=0, \
+        label={'label': label_name, 'style': {'color': '#C4CDD5'}}, \
+        max=max_value, \
+        min=0, \
+    )
+
+app.layout = html.Div([
+    html.Div(children=[
+        dcc.Interval(id="default-interval", interval=1 * 2000, n_intervals=0),
+
+        html.Div(children=[
+
+            dt.DataTable(
+                id='table-info',
+                columns=[{'id': c.lower(), 'name': c, 'editable': False}
+                         for c in ['Key', 'Value']],
+            )
+
+        ], style={'display': 'flex', 'flex-direction': 'row'}),
+
+        html.Div(children=[
+            dcc.Graph(
+                id='piechart-flows',
+                config={
+                    'displayModeBar': False,
+                },
+            ),
+        ], style={'padding': 10, 'flex': 1}),
+
+        html.Div(children=[
+            dcc.Graph(
+                id='piechart-midstream-flows',
+                config={
+                    'displayModeBar': False,
+                },
+            ),
+        ], style={'padding': 10, 'flex': 1}),
+
+        html.Div(children=[
+            dcc.Graph(
+                id='piechart-risky-flows',
+                config={
+                    'displayModeBar': False,
+                },
+            ),
+        ], style={'padding': 10, 'flex': 1}),
+    ], style=generate_box()),
+
+    html.Div(children=[
+        dcc.Interval(id="graph-interval", interval=4 * 1000, n_intervals=0),
+        dcc.Store(id="graph-traces"),
+
+        html.Div(children=[
+            dcc.Graph(
+                id="graph-flows",
+                config={
+                    'displayModeBar': False,
+                },
+                style={'height':'60vh'},
+            ),
+        ], style={'padding': 10, 'flex': 1})
+    ], style=generate_box()),
+])
+
+def build_gauge(key, max_value=100):
+    gauge_max = int(max(max_value,
+                        shared_flow_dict[key]))
+    grad_green  = [0,                    int(gauge_max * 1/3)]
+    grad_yellow = [int(gauge_max * 1/3), int(gauge_max * 2/3)]
+    grad_red    = [int(gauge_max * 2/3), gauge_max]
+
+    grad_dict   = \
+        { \
+            "gradient":True, \
+            "ranges":{ \
+                "green":grad_green, \
+                "yellow":grad_yellow, \
+                "red":grad_red \
+            } \
+        }
+
+    return shared_flow_dict[key], gauge_max, grad_dict
+
+def build_piechart(labels, values):
+    lay = dict(
+        plot_bgcolor = '#082255',
+        paper_bgcolor = '#082255',
+        font={"color": "#fff"},
+        autosize=True,
+        height=250,
+        margin = {'autoexpand': False, 'b': 0, 'l': 0, 'r': 0, 't': 0, 'pad': 0},
+        width = 500,
+        uniformtext_minsize = 12,
+        uniformtext_mode = 'hide',
+    )
+
+    return go.Figure(layout=lay, data=[go.Pie(labels=labels, values=values, textinfo='percent', textposition='inside')])
+
+def prettifyBytes(bytes_received):
+    size_names = ['B', 'KB', 'MB', 'GB', 'TB']
+    if bytes_received == 0:
+        i = 0
+    else:
+        i = min(int(math.floor(math.log(bytes_received, 1024))), len(size_names) - 1)
+    p = math.pow(1024, i)
+    s = round(bytes_received / p, 2)
+    return '{:.2f} {}'.format(s, size_names[i])
+
+@app.callback(output=[Output('table-info', 'data'),
+                      Output('piechart-flows', 'figure'),
+                      Output('piechart-midstream-flows', 'figure'),
+                      Output('piechart-risky-flows', 'figure')],
+
+              inputs=[Input('default-interval', 'n_intervals')])
+def update_led_gauge(n):
+    return [[{'key': 'Total JSON Events',        'value': shared_flow_dict['total-events']},
+             {'key': 'Total JSON Bytes',         'value': prettifyBytes(shared_flow_dict['total-bytes'])},
+             {'key': 'Total Flows',              'value': shared_flow_dict['total-flows']},
+             {'key': 'Total Risky Flows',        'value': shared_flow_dict['total-risky-flows']},
+             {'key': 'Total Midstream Flows',    'value': shared_flow_dict['total-midstream-flows']},
+             {'key': 'Total Guessed Flows',      'value': shared_flow_dict['total-guessed-flows']},
+             {'key': 'Total Not Detected Flows', 'value': shared_flow_dict['total-not-detected-flows']}],
+            build_piechart(['Detected', 'Guessed', 'Undetected', 'Unclassified'],
+                           [shared_flow_dict['current-detected-flows'],
+                            shared_flow_dict['current-guessed-flows'],
+                            shared_flow_dict['current-not-detected-flows'],
+                            shared_flow_dict['current-flows']
+                                - shared_flow_dict['current-detected-flows']
+                                - shared_flow_dict['current-guessed-flows']
+                                - shared_flow_dict['current-not-detected-flows']]),
+            build_piechart(['Midstream', 'Not Midstream'],
+                           [shared_flow_dict['current-midstream-flows'],
+                            shared_flow_dict['current-flows'] -
+                            shared_flow_dict['current-midstream-flows']]),
+            build_piechart(['Risky', 'Not Risky'],
+                           [shared_flow_dict['current-risky-flows'],
+                            shared_flow_dict['current-flows'] -
+                            shared_flow_dict['current-risky-flows']])]
+
+@app.callback(output=[Output('graph-flows', 'figure'),
+                      Output('graph-traces', 'data')],
+              inputs=[Input('graph-interval', 'n_intervals'),
+                      Input('graph-interval', 'interval')],
+              state=[State('graph-traces', 'data')])
+def update_graph(n, i, traces):
+    if traces is None:
+        traces = ([], [], [], [], [], [])
+
+    max_bins = 50
+
+    traces[0].append(shared_flow_dict['current-flows'])
+    traces[1].append(shared_flow_dict['current-risky-flows'])
+    traces[2].append(shared_flow_dict['current-midstream-flows'])
+    traces[3].append(shared_flow_dict['current-guessed-flows'])
+    traces[4].append(shared_flow_dict['current-not-detected-flows'])
+    traces[5].append(shared_flow_dict['current-flows']
+                        - shared_flow_dict['current-detected-flows']
+                        - shared_flow_dict['current-guessed-flows']
+                        - shared_flow_dict['current-not-detected-flows'])
+    if len(traces[0]) > max_bins:
+        traces[0] = traces[0][1:]
+        traces[1] = traces[1][1:]
+        traces[2] = traces[2][1:]
+        traces[3] = traces[3][1:]
+        traces[4] = traces[4][1:]
+        traces[5] = traces[5][1:]
+
+    i /= 1000.0
+    x = list(range(max(n - max_bins, 0) * int(i), n * int(i), max(int(i), 1)))
+
+    lay = dict(
+        plot_bgcolor = '#082255',
+        paper_bgcolor = '#082255',
+        font={"color": "#fff"},
+        xaxis = {
+            'title': 'Time (sec)',
+            "showgrid": False,
+            "showline": False,
+            "fixedrange": True,
+            "tickmode": 'linear',
+            "dtick": i,
+        },
+        yaxis = {
+            'title': 'Flow Count',
+            "showgrid": False,
+            "showline": False,
+            "zeroline": False,
+            "fixedrange": True,
+            "tickmode": 'linear',
+            "dtick": 10,
+        },
+        autosize=True,
+        bargap=0.01,
+        bargroupgap=0,
+        hovermode="closest",
+        margin = {'b': 0, 'l': 0, 'r': 0, 't': 0, 'pad': 0},
+        legend = {'borderwidth': 0},
+    )
+
+    fig = go.Figure(layout=lay)
+    fig.update_xaxes(showgrid=True, gridwidth=1, gridcolor='#007ACE', zeroline=True, zerolinewidth=1)
+    fig.update_yaxes(showgrid=True, gridwidth=1, gridcolor='#007ACE', zeroline=True, zerolinewidth=1)
+    fig.add_trace(go.Scatter(
+        x=x,
+        y=traces[0],
+        name='Current Active Flows',
+    ))
+    fig.add_trace(go.Scatter(
+        x=x,
+        y=traces[1],
+        name='Current Risky Flows',
+    ))
+    fig.add_trace(go.Scatter(
+        x=x,
+        y=traces[2],
+        name='Current Midstream Flows',
+    ))
+    fig.add_trace(go.Scatter(
+        x=x,
+        y=traces[3],
+        name='Current Guessed Flows',
+    ))
+    fig.add_trace(go.Scatter(
+        x=x,
+        y=traces[4],
+        name='Current Not Detected Flows',
+    ))
+    fig.add_trace(go.Scatter(
+        x=x,
+        y=traces[5],
+        name='Current Unclassified Flows',
+    ))
+
+    return [fig, traces]
+
+def web_worker(mp_shared_flow_dict):
+    global shared_flow_dict
+
+    shared_flow_dict = mp_shared_flow_dict
+
+    app.run_server(debug=False)
diff --git a/examples/py-flow-info/flow-info.py b/examples/py-flow-info/flow-info.py
index 1f25cea55..42f70d813 100755
--- a/examples/py-flow-info/flow-info.py
+++ b/examples/py-flow-info/flow-info.py
@@ -72,7 +72,11 @@ class Stats:
         set_attr_from_dict(current_flow, json_dict, 'flow_event_name', '')
         set_attr_if_not_set(current_flow, 'guessed', False)
         set_attr_if_not_set(current_flow, 'not_detected', False)
-        if current_flow.flow_event_name == 'guessed':
+
+        if current_flow.flow_event_name == 'detected' or \
+           current_flow.flow_event_name == 'detection-update':
+            current_flow.guessed = False
+        elif current_flow.flow_event_name == 'guessed':
             current_flow.guessed = True
         elif current_flow.flow_event_name == 'not-detected':
             current_flow.not_detected = True