host_statistics.py 文件源码

def get_host_distinct_ports():
    """
    Gets flows, packet and bytes time series for a given host

    Returns: JSON with status "ok" or "error" and requested data.

    """
    # Check login
    if not session.logged:
        json_response = '{"status": "Error", "data": "You must be logged!"}'
        return json_response

    # Check mandatory inputs
    if not (request.get_vars.beginning and request.get_vars.end and request.get_vars.aggregation and request.get_vars.host_ip):
        json_response = '{"status": "Error", "data": "Some mandatory argument is missing!"}'
        return json_response

    # Parse inputs and set correct format
    beginning = escape(request.get_vars.beginning)
    end = escape(request.get_vars.end)
    aggregation = escape(request.get_vars.aggregation)
    host_ip = escape(request.get_vars.host_ip)

    try:
        # Elastic query
        client = elasticsearch.Elasticsearch(
            [{'host': myconf.get('consumer.hostname'), 'port': myconf.get('consumer.port')}])
        elastic_bool = []
        elastic_bool.append({'range': {'@timestamp': {'gte': beginning, 'lte': end}}})
        elastic_bool.append({'term': {'src_ip': host_ip}})

        qx = Q({'bool': {'must': elastic_bool}})
        s = Search(using=client, index='_all').query(qx)
        s.aggs.bucket('by_time', 'date_histogram', field='@timestamp', interval=aggregation) \
              .metric('dport_avg', 'avg', field='stats.dport_count') \
              .metric('dport_max', 'max', field='stats.dport_count') \
              .metric('dport_min', 'min', field='stats.dport_count')

        result = s.execute()

        data_avg = []
        data_min_max = []
        data_max = []
        data_min = []
        for record in result.aggregations.by_time.buckets:
            timestamp = record.key
            maximum = round(record.dport_max.value, 2) if record.dport_max.value else None
            minimum = round(record.dport_min.value, 2) if record.dport_min.value else None
            data_avg.append([timestamp,round(record.dport_avg.value,2) if record.dport_avg.value else None])
            data_min_max.append([timestamp,[minimum, maximum ]])
            data_max.append(maximum)
            data_min.append(minimum)

        json_response = {"status": "Ok", "data":{ "data_avg": data_avg, "data_min_max": data_min_max, "data_min": data_min, "data_max": data_max}}
        return (json.dumps(json_response))

    except Exception as e:
        json_response = '{"status": "Error", "data": "Elasticsearch query exception: ' + escape(str(e)) + '"}'
        return json_response