Merge pull request #606 from landrushka/master

added boolean 'labeled', 'prefix_measurement' options to Influx plugi…
yandex · Jun 5, 2018 · 40ca96c · 40ca96c
2 parents 35c1035 + add40ab
commit 40ca96c
Show file tree

Hide file tree

Showing 3 changed files with 70 additions and 2 deletions.
diff --git a/yandextank/plugins/Influx/config/schema.yaml b/yandextank/plugins/Influx/config/schema.yaml
@@ -24,4 +24,10 @@ grafana_dashboard:
   type: string
 chunk_size:
   default: 500000
-  type: integer
+  type: integer
+labeled:
+  default: false
+  type: boolean
+prefix_measurement:
+  default: ""
+  type: string
diff --git a/yandextank/plugins/Influx/decoder.py b/yandextank/plugins/Influx/decoder.py
@@ -44,6 +44,63 @@ def decode_monitoring(self, data):
                 }]
         return points
 
+    def decode_aggregate_labeled(self, data, stat, prefix):
+        timestamp = int(data["ts"])
+        points = []
+        for key, value in data["tagged"].items():
+            points += [{
+                "measurement": prefix + "overall_quantiles",
+                "tags": {
+                    "tank": self.tank_tag,
+                    "uuid": self.uuid,
+                    "label": key
+                },
+                "time": timestamp,
+                "fields": {  # quantiles
+                    'q' + str(q): value / 1000.0
+                    for q, value in zip(value["interval_real"]["q"]["q"],
+                                        value["interval_real"]["q"]["value"])
+                },
+            }, {
+                "measurement": prefix + "overall_meta",
+                "tags": {
+                    "tank": self.tank_tag,
+                    "uuid": self.uuid,
+                    "label": key,
+                },
+                "time": timestamp,
+                "fields": {
+                    "active_threads": stat["metrics"]["instances"],
+                    "RPS": value["interval_real"]["len"],
+                    "planned_requests": float(stat["metrics"]["reqps"]),
+                },
+            }, {
+                "measurement": prefix + "net_codes",
+                "tags": {
+                    "tank": self.tank_tag,
+                    "uuid": self.uuid,
+                    "label": key
+                },
+                "time": timestamp,
+                "fields": {
+                    str(code): int(cnt)
+                    for code, cnt in value["net_code"]["count"].items()
+                },
+            }, {
+                "measurement": prefix + "proto_codes",
+                "tags": {
+                    "tank": self.tank_tag,
+                    "uuid": self.uuid,
+                },
+                "time": timestamp,
+                "fields": {
+                    str(code): int(cnt)
+                    for code, cnt in value["proto_code"]["count"].items()
+                },
+            },
+            ]
+        return points
+
     def decode_aggregate(self, data, stat):
         timestamp = int(data["ts"])
         points = [

diff --git a/yandextank/plugins/Influx/plugin.py b/yandextank/plugins/Influx/plugin.py
@@ -45,6 +45,8 @@ def __init__(self, core, cfg):
             password=self.get_option("password"),
             database=self.get_option("database"),
         )
+        self.labeled = self.get_option("labeled")
+        self.prefix_measurement = self.get_option("prefix_measurement")
         grafana_root = self.get_option("grafana_root")
         grafana_dashboard = self.get_option("grafana_dashboard")
         uuid = str(uuid4())
@@ -70,7 +72,10 @@ def prepare_test(self):
 
     def on_aggregated_data(self, data, stats):
         if self.client:
-            points = self.decoder.decode_aggregate(data, stats)
+            if self.labeled:
+                points = self.decoder.decode_aggregate_labeled(data, stats, self.prefix_measurement)
+            else:
+                points = self.decoder.decode_aggregate(data, stats)
             self.client.write_points(points, 's')
 
     def monitoring_data(self, data_list):