Histostat metrics support

.github/workflows/python-app.yml

@@ -31,7 +31,8 @@ jobs:
         # stop the build if there are Python syntax errors or undefined names
         flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+        # Black declares E203 is not PEP8 compliant: https://github.com/psf/black/issues/315#issuecomment-395457972
+        flake8 . --count --ignore=E203 --exit-zero --max-complexity=10 --max-line-length=127 --statistics
     - name: Test with pytest
       run: |
         poetry run pytest tests

hunter/config.py

@@ -1,9 +1,9 @@
 import os
 from dataclasses import dataclass
-from typing import Dict, List
+from pathlib import Path
+from typing import Dict, List, Optional
 
 from expandvars import expandvars
-from pathlib import Path
 from ruamel.yaml import YAML
 
 from hunter.grafana import GrafanaConfig
@@ -15,8 +15,8 @@
 
 @dataclass
 class Config:
-    graphite: GraphiteConfig
-    grafana: GrafanaConfig
+    graphite: Optional[GraphiteConfig]
+    grafana: Optional[GrafanaConfig]
     tests: Dict[str, TestConfig]
     test_groups: Dict[str, List[TestConfig]]
     slack: SlackConfig
@@ -30,14 +30,14 @@ class ConfigError(Exception):
 def load_templates(config: Dict) -> Dict[str, Dict]:
     templates = config.get("templates", {})
     if not isinstance(templates, Dict):
-        raise ConfigError(f"Property `templates` is not a dictionary")
+        raise ConfigError("Property `templates` is not a dictionary")
     return templates
 
 
 def load_tests(config: Dict, templates: Dict) -> Dict[str, TestConfig]:
     tests = config.get("tests", {})
     if not isinstance(tests, Dict):
-        raise ConfigError(f"Property `tests` is not a dictionary")
+        raise ConfigError("Property `tests` is not a dictionary")
 
     result = {}
     for (test_name, test_config) in tests.items():
@@ -57,7 +57,7 @@ def load_tests(config: Dict, templates: Dict) -> Dict[str, TestConfig]:
 def load_test_groups(config: Dict, tests: Dict[str, TestConfig]) -> Dict[str, List[TestConfig]]:
     groups = config.get("test_groups", {})
     if not isinstance(groups, Dict):
-        raise ConfigError(f"Property `test_groups` is not a dictionary")
+        raise ConfigError("Property `test_groups` is not a dictionary")
 
     result = {}
     for (group_name, test_names) in groups.items():
@@ -85,13 +85,22 @@ def load_config_from(config_file: Path) -> Config:
         if Grafana configs not explicitly set in yaml file, default to same as Graphite 
         server at port 3000
         """
-        if config["graphite"]["url"] is None:
-            raise ValueError("graphite.url")
-        if config.get("grafana") is None:
-            config["gafana"] = {}
-            config["grafana"]["url"] = f"{config['graphite']['url'].strip('/')}:3000/"
-            config["grafana"]["user"] = os.environ.get("GRAFANA_USER", "admin")
-            config["grafana"]["password"] = os.environ.get("GRAFANA_PASSWORD", "admin")
+        graphite_config = None
+        grafana_config = None
+        if "graphite" in config:
+            if "url" not in config["graphite"]:
+                raise ValueError("graphite.url")
+            graphite_config = GraphiteConfig(url=config["graphite"]["url"])
+            if config.get("grafana") is None:
+                config["grafana"] = {}
+                config["grafana"]["url"] = f"{config['graphite']['url'].strip('/')}:3000/"
+                config["grafana"]["user"] = os.environ.get("GRAFANA_USER", "admin")
+                config["grafana"]["password"] = os.environ.get("GRAFANA_PASSWORD", "admin")
+            grafana_config = GrafanaConfig(
+                url=config["grafana"]["url"],
+                user=config["grafana"]["user"],
+                password=config["grafana"]["password"],
+            )
 
         slack_config = None
         if config.get("slack") is not None:
@@ -106,12 +115,8 @@ def load_config_from(config_file: Path) -> Config:
         groups = load_test_groups(config, tests)
 
         return Config(
-            graphite=GraphiteConfig(url=config["graphite"]["url"]),
-            grafana=GrafanaConfig(
-                url=config["grafana"]["url"],
-                user=config["grafana"]["user"],
-                password=config["grafana"]["password"],
-            ),
+            graphite=graphite_config,
+            grafana=grafana_config,
             slack=slack_config,
             tests=tests,
             test_groups=groups,

hunter/csv_options.py

@@ -1,7 +1,6 @@
 import enum
 
 from dataclasses import dataclass
-from typing import Optional
 
 
 @dataclass

hunter/data_selector.py

@@ -1,7 +1,7 @@
 import sys
 from dataclasses import dataclass
-from typing import List, Optional
 from datetime import datetime, timedelta
+from typing import List, Optional
 
 import pytz
 

hunter/grafana.py

@@ -1,13 +1,10 @@
-import re
+from dataclasses import asdict, dataclass
 from datetime import datetime
+from typing import List, Optional
 
 import requests
-
-from dataclasses import asdict, dataclass
-
 from pytz import UTC
 from requests.exceptions import HTTPError
-from typing import List, Optional
 
 
 @dataclass

hunter/graphite.py

@@ -1,10 +1,9 @@
 import ast
 import json
 import urllib.request
-
 from dataclasses import dataclass
 from datetime import datetime
-from logging import info, warning
+from logging import info
 from typing import Dict, List, Optional, Iterable
 
 from hunter.data_selector import DataSelector

hunter/importer.py

@@ -1,26 +1,29 @@
 import csv
 from collections import OrderedDict
-
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import List, Optional, Dict
 
 from hunter.config import Config
-from hunter.csv_options import CsvColumnType, CsvOptions
 from hunter.data_selector import DataSelector
 from hunter.graphite import DataPoint, Graphite, GraphiteError
 from hunter.series import Series, Metric
-from hunter.test_config import CsvTestConfig, TestConfig, GraphiteTestConfig, CsvMetric
+from hunter.test_config import (
+    CsvTestConfig,
+    TestConfig,
+    GraphiteTestConfig,
+    CsvMetric,
+    HistoStatTestConfig,
+)
 from hunter.util import (
     merge_sorted,
     parse_datetime,
     DateFormatError,
     format_timestamp,
     resolution,
     round,
-    is_float,
-    is_datetime,
 )
 
 
@@ -107,7 +110,7 @@ def fetch_data(self, test: TestConfig, selector: DataSelector = DataSelector())
                 raise DataImportError(f"No timeseries found in Graphite for test {test.name}.")
 
             times = [[x.time for x in series.points] for series in graphite_result]
-            time: List[int] = merge_sorted(times)[-selector.last_n_points:]
+            time: List[int] = merge_sorted(times)[-selector.last_n_points :]
 
             def column(series: List[DataPoint]) -> List[float]:
                 value_by_time = dict([(x.time, x.value) for x in series])
@@ -281,15 +284,15 @@ def fetch_data(self, test_conf: TestConfig, selector: DataSelector = DataSelecto
                 metrics = {m.name: Metric(m.direction, m.scale) for m in metrics.values()}
 
                 # Leave last n points:
-                time = time[-selector.last_n_points:]
+                time = time[-selector.last_n_points :]
                 tmp = data
                 data = {}
                 for k, v in tmp.items():
-                    data[k] = v[-selector.last_n_points:]
+                    data[k] = v[-selector.last_n_points :]
                 tmp = attributes
                 attributes = {}
                 for k, v in tmp.items():
-                    attributes[k] = v[-selector.last_n_points:]
+                    attributes[k] = v[-selector.last_n_points :]
 
                 return Series(
                     test_conf.name,
@@ -314,15 +317,131 @@ def fetch_all_metric_names(self, test_conf: CsvTestConfig) -> List[str]:
         return [m for m in test_conf.metrics.keys()]
 
 
+class HistoStatImporter(Importer):
+
+    __TAG_METRICS = {
+        "count": {"direction": 1, "scale": "1", "col": 3},
+        "min": {"direction": -1, "scale": "1.0e-6", "col": 4},
+        "p25": {"direction": -1, "scale": "1.0e-6", "col": 5},
+        "p50": {"direction": -1, "scale": "1.0e-6", "col": 6},
+        "p75": {"direction": -1, "scale": "1.0e-6", "col": 7},
+        "p90": {"direction": -1, "scale": "1.0e-6", "col": 8},
+        "p95": {"direction": -1, "scale": "1.0e-6", "col": 9},
+        "p98": {"direction": -1, "scale": "1.0e-6", "col": 10},
+        "p99": {"direction": -1, "scale": "1.0e-6", "col": 11},
+        "p999": {"direction": -1, "scale": "1.0e-6", "col": 12},
+        "p9999": {"direction": -1, "scale": "1.0e-6", "col": 13},
+        "max": {"direction": -1, "scale": "1.0e-6", "col": 14},
+    }
+
+    @contextmanager
+    def __csv_reader(self, test: HistoStatTestConfig):
+        with open(Path(test.file), newline="") as csv_file:
+            yield csv.reader(csv_file)
+
+    @staticmethod
+    def __parse_tag(tag: str):
+        return tag.split("=")[1]
+
+    def __get_tags(self, test: HistoStatTestConfig) -> List[str]:
+        tags = set()
+        with self.__csv_reader(test) as reader:
+            for row in reader:
+                if row[0].startswith("#"):
+                    continue
+                tag = self.__parse_tag(row[0])
+                if tag in tags:
+                    break
+                tags.add(tag)
+        return list(tags)
+
+    @staticmethod
+    def __metric_from_components(tag, tag_metric):
+        return f"{tag}.{tag_metric}"
+
+    @staticmethod
+    def __convert_floating_point_millisecond(fpm: str) -> int:  # to epoch seconds
+        return int(float(fpm) * 1000) // 1000
+
+    def fetch_data(
+        self, test: HistoStatTestConfig, selector: DataSelector = DataSelector()
+    ) -> Series:
+        def selected(metric_name):
+            return metric_name in selector.metrics if selector.metrics is not None else True
+
+        metrics = {}
+        tag_count = 0
+        for tag in self.__get_tags(test):
+            tag_count += 1
+            for tag_metric, attrs in self.__TAG_METRICS.items():
+                if selected(self.__metric_from_components(tag, tag_metric)):
+                    metrics[self.__metric_from_components(tag, tag_metric)] = Metric(
+                        attrs["direction"], attrs["scale"]
+                    )
+
+        data = {k: [] for k in metrics.keys()}
+        time = []
+        with self.__csv_reader(test) as reader:
+            start_time = None
+            for row in reader:
+                if not row[0].startswith("#"):
+                    break
+                if "StartTime" in row[0]:
+                    parts = row[0].split(" ")
+                    start_time = self.__convert_floating_point_millisecond(parts[1])
+
+            if not start_time:
+                raise DataImportError("No Start Time specified in HistoStat CSV comment")
+
+            # Last iteration of row is the first non-comment row. Parse it now.
+            tag_interval = 0
+            while row:
+                if tag_interval % tag_count == 0:
+                    # Introduces a slight inaccuracy - each tag can report its interval start time
+                    # with some millisecond difference. Choosing a single tag interval allows us
+                    # to maintain the 'indexed by a single time variable' contract required by
+                    # Series, but the time reported for almost all metrics will be _slightly_ off.
+                    time.append(self.__convert_floating_point_millisecond(row[1]) + start_time)
+                tag_interval += 1
+                tag = self.__parse_tag(row[0])
+                for tag_metric, attrs in self.__TAG_METRICS.items():
+                    if selected(self.__metric_from_components(tag, tag_metric)):
+                        data[self.__metric_from_components(tag, tag_metric)].append(
+                            float(row[attrs["col"]])
+                        )
+                try:
+                    row = next(reader)
+                except StopIteration:
+                    row = None
+
+        # Leave last n points:
+        time = time[-selector.last_n_points :]
+        tmp = data
+        data = {}
+        for k, v in tmp.items():
+            data[k] = v[-selector.last_n_points :]
+
+        return Series(test.name, None, time, metrics, data, dict())
+
+    def fetch_all_metric_names(self, test: HistoStatTestConfig) -> List[str]:
+        metric_names = []
+        for tag in self.__get_tags(test):
+            for tag_metric in self.__TAG_METRICS.keys():
+                metric_names.append(self.__metric_from_components(tag, tag_metric))
+        return metric_names
+
+
 class Importers:
     __config: Config
     __csv_importer: Optional[CsvImporter]
     __graphite_importer: Optional[GraphiteImporter]
+    __histostat_importer: Optional[HistoStatImporter]
 
     def __init__(self, config: Config):
         self.__config = config
         self.__csv_importer = None
         self.__graphite_importer = None
+        self.__histostat_importer = None
 
     def csv_importer(self) -> CsvImporter:
         if self.__csv_importer is None:
@@ -334,10 +453,17 @@ def graphite_importer(self) -> GraphiteImporter:
             self.__graphite_importer = GraphiteImporter(Graphite(self.__config.graphite))
         return self.__graphite_importer
 
+    def histostat_importer(self) -> HistoStatImporter:
+        if self.__histostat_importer is None:
+            self.__histostat_importer = HistoStatImporter()
+        return self.__histostat_importer
+
     def get(self, test: TestConfig) -> Importer:
         if isinstance(test, CsvTestConfig):
             return self.csv_importer()
         elif isinstance(test, GraphiteTestConfig):
             return self.graphite_importer()
+        elif isinstance(test, HistoStatTestConfig):
+            return self.histostat_importer()
         else:
             raise ValueError(f"Unsupported test type {type(test)}")