Use a Go tool to improve importing performance (#37)

Author: AstroProfundis

.travis.yml

@@ -5,6 +5,7 @@ language: go
 go:
   - "1.9.x"
   - "1.10.x"
+  - "1.11.x"
   - master
 
 matrix:

Gopkg.lock

@@ -31,7 +31,15 @@
 
 [[constraint]]
   name = "github.com/shirou/gopsutil"
-  version = "2.18.06"
+  version = "2.18.07"
+
+[[constraint]]
+  name = "github.com/influxdata/influxdb"
+  version = "v1.6.2"
+
+[[constraint]]
+  branch = "master"
+  name = "github.com/prometheus/common"
 
 [prune]
   go-tests = true

Gopkg.toml

@@ -1,16 +1,17 @@
 ### Makefile for tidb-insight
-.PHONY: collector vmtouch
+.PHONY: collector tools
 
-default: collector vmtouch
+default: collector tools
 
-debug: collector vmtouch
+debug: collector tools
 
 all: default
 
 collector:
 	$(MAKE) -C collector $(MAKECMDGOALS)
 
-vmtouch:
+tools:
+	$(MAKE) -C tools $(MAKECMDGOALS)
 	$(MAKE) -C tools/vmtouch $(MAKECMDGOALS)
 	install -Dm755 tools/vmtouch/vmtouch bin/
 

Makefile

@@ -25,7 +25,7 @@ def save_sysconf(self):
             fileopt.write_file(os.path.join(
                 self.outdir, "sysctl.conf"), stdout)
         if stderr:
-            logging.warn("Reading limits.conf returned error: %s" % stderr)
+            logging.warning("Reading limits.conf returned error: %s" % stderr)
 
     def find_tidb_configfiles(self, cmdline=None):
         cmd_opts = util.parse_cmdline(cmdline)

file/configfiles.py

@@ -238,7 +238,7 @@ def get_lsof_tidb(self):
     def save_logfiles(self, args):
         # reading logs requires root priviledge
         if not util.is_root_privilege():
-            logging.warn("It's required to read logs with root priviledge.")
+            logging.warning("It's required to read logs with root priviledge.")
             # return
 
         self.insight_logfiles = logfiles.InsightLogFiles(

insight.py

@@ -5,13 +5,11 @@
 # for futher analysis.
 
 import datetime
-import influxdb
 import json
 import logging
 import os
 import random
 import string
-import zlib
 
 from utils import fileopt
 from utils import util
@@ -41,7 +39,26 @@ def unique_dbname(self):
 
         return '_'.join(dbname)
 
-    def load_dump(self):
+    def exec_importer(self, file=None, chunk_size=2000):
+        if not file:
+            logging.fatal("No file specified.")
+            return (None, "No metric dump file specified to load.")
+        base_dir = os.path.join(util.pwd(), "../")
+        importer = os.path.join(base_dir, "bin/prom2influx")
+        cmd = [importer,
+               "-db", self.db_name,
+               "-host", self.host,
+               "-port", "%s" % self.port,
+               "-chunk", "%s" % chunk_size,  # chunk size of one write request
+               "-file", file
+               ]
+        logging.debug("Running cmd: %s" % ' '.join(cmd))
+        return util.run_cmd(cmd)
+
+    def run_importing(self):
+        logging.info("Metrics will be imported to database '%s'." %
+                     self.db_name)
+
         def file_list(dir=None):
             f_list = []
             for file in fileopt.list_dir(dir):
@@ -52,58 +69,17 @@ def file_list(dir=None):
             return f_list
 
         for file in file_list(self.datadir):
-            if file.endswith('.json'):
-                raw = fileopt.read_file(file)
-            elif file.endswith('.dat'):
-                raw = zlib.decompress(fileopt.read_file(file, 'rb'))
-            else:
-                logging.debug("Skipped unrecorgnized file '%s'" % file)
+            # all dumped files are in 'prometheus' sub-directory
+            if not file or not file.endswith('.json') or 'prometheus' not in file:
                 continue
-            yield json.loads(raw)
-
-    def build_series(self):
-        def format_prom_metric(key=None):
-            points = []
-            point = {'fields': {}}
-            # build point header
-            for metric in key:
-                point['measurement'] = metric['metric']['__name__']
-                point['tags'] = {
-                    'cluster': self.db_name,
-                    'monitor': 'prometheus',
-                }
-                for k, v in metric['metric'].items():
-                    point['tags'][k] = v
-                # build point values
-                for value in metric['values']:
-                    point['time'] = datetime.datetime.utcfromtimestamp(
-                        value[0]).strftime('%Y-%m-%dT%H:%M:%SZ')
-                    try:
-                        point['fields']['value'] = float(value[1])
-                    except ValueError:
-                        point['fields']['value'] = value[1]
-                    points.append(point.copy())
-            return points
-
-        for key in self.load_dump():
-            yield format_prom_metric(key)
-
-    def write2influxdb(self):
-        client = influxdb.InfluxDBClient(
-            host=self.host, port=self.port, username=self.user, password=self.passwd,
-            database=self.db_name, timeout=30)
-        # create_database has no effect if the database already exist
-        client.create_database(self.db_name)
-        logging.info("Metrics will be imported to database '%s'." %
-                     self.db_name)
-
-        for series in self.build_series():
-            try:
-                client.write_points(series, batch_size=2000)
-            except influxdb.exceptions.InfluxDBClientError as e:
-                logging.warn(
-                    "Write error for key '%s', data may be empty." % series[0]['measurement'])
-                logging.debug(e)
-
-    def run_importing(self):
-        self.write2influxdb()
+            stderr = self.exec_importer(file)[1]
+            if stderr and "Request Entity Too Large" in stderr.decode('utf-8'):
+                logging.info("Write to DB failed, retry for once...")
+                retry_stderr = self.exec_importer(file, chunk_size=100)[1]
+                if not retry_stderr:
+                    logging.info("Retry succeeded.")
+                else:
+                    logging.warning("Retry failed, stderr is: '%s'" %
+                                    retry_stderr)
+            elif stderr:
+                logging.warning(stderr)

metric/importer/prometheus.py

@@ -6,7 +6,6 @@
 import json
 import logging
 import os
-import zlib
 
 from metric.base import MetricBase
 from utils import fileopt
@@ -36,7 +35,7 @@ def get_label_names(self):
 
     def run_collecting(self):
         if self.resolution < 15.0:
-            logging.warn(
+            logging.warning(
                 "Sampling resolution < 15s don't increase accuracy but data size.")
         for metric in self.get_label_names():
             url = '%s/query_range?query=%s&start=%s&end=%s&step=%s' % (
@@ -46,14 +45,8 @@ def run_collecting(self):
                 logging.info("Error querying for key '%s'." % metric)
                 logging.debug("Output is:\n%s" % matrix)
                 continue
-            if self.options.compress:
-                metric_filename = '%s_%s_to_%s_%ss.dat' % (
-                    metric, self.start_time, self.end_time, self.resolution)
-                fileopt.write_file(os.path.join(self.outdir, metric_filename), zlib.compress(
-                    json.dumps(matrix['data']['result'])))
-            else:
-                metric_filename = '%s_%s_to_%s_%ss.json' % (
-                    metric, self.start_time, self.end_time, self.resolution)
-                fileopt.write_file(os.path.join(
-                    self.outdir, metric_filename), json.dumps(matrix['data']['result']))
+            metric_filename = '%s_%s_to_%s_%ss.json' % (
+                metric, self.start_time, self.end_time, self.resolution)
+            fileopt.write_file(os.path.join(
+                self.outdir, metric_filename), json.dumps(matrix['data']['result']))
             logging.debug("Saved data for key '%s'." % metric)

metric/prometheus.py

@@ -11,7 +11,7 @@ else
 fi
 RELPATH=${PKGNAME}-${RELVER}
 
-GO_RELEASE_BIN=go1.10.3.linux-amd64
+GO_RELEASE_BIN=go1.11.linux-amd64
 
 BUILD_ROOT="`pwd`/.build"
 mkdir -p ${BUILD_ROOT}
@@ -38,6 +38,8 @@ ln -sfv vendor src
 # compile a static binary
 cd ${BUILD_ROOT}/${PKGNAME}/collector/
 GOBIN=${GOROOT}/bin/go make static || exit 1
+cd ${BUILD_ROOT}/${PKGNAME}/tools/
+GOBIN=${GOROOT}/bin/go make static || exit 1
 
 # compile other tools
 cd ${BUILD_ROOT}/${PKGNAME}/tools/vmtouch

package.sh

@@ -82,13 +82,13 @@ def run_collecting(self):
                     fileopt.write_file(
                         path.join(self.outdir, "%s.stdout" % pname), stdout)
                 if stderr:
-                    logging.warn(
+                    logging.warning(
                         "Command '%s' returned error: %s" % (cmd, stderr))
                 if self.options.archive:
                     cmd = self.build_archive_cmd(pid, pname)
                     stdout, stderr = util.run_cmd(cmd)
                     if stderr:
-                        logging.warn(
+                        logging.warning(
                             "Command '%s' returned error: %s" % (cmd, stderr))
         else:
             # perf the entire system
@@ -98,10 +98,11 @@ def run_collecting(self):
                 fileopt.write_file(
                     path.join(self.outdir, "perf.stdout"), stdout)
             if stderr:
-                logging.warn("Command '%s' returned error: %s" % (cmd, stderr))
+                logging.warning("Command '%s' returned error: %s" %
+                                (cmd, stderr))
             if self.options.archive:
                 cmd = self.build_archive_cmd()
                 stdout, stderr = util.run_cmd(cmd)
                 if stderr:
-                    logging.warn(
+                    logging.warning(
                         "Command '%s' returned error: %s" % (cmd, stderr))