Support option to set yaml config folder (#62)

* Bumped up number of retries

* Set timeout while sending data over http/s

* Thread scheduling improvements
Do not starve other collectors if one collector is pushing too much data
Do not sleep when there is data to be data to be read/sent

* mentric printer thread should exit if the main thread exits

* Added flush method to stdout/stderr streams

* Added support for flushBeforeExit option so that tests wait for all data
Assertions are to be applied only after all data is received

* Converted flushBeforeExit option to function to make pylint happy
Pylint was failing with following error even though we had hasattr check
E: 41,88: Module 'collectors.etc.udp_bridge_conf' has no 'flushBeforeExit' member (no-member)

* Converted flushBeforeExit option to function to make pylint happy
Pylint was failing with following error even though we had hasattr check
E: 41,88: Module 'collectors.etc.udp_bridge_conf' has no 'flushBeforeExit' member (no-member)

* Codacy love

* Support an option to set yaml config folder
Allow running collectors for diff tenants (different xcollector.yml files) from same installation folder

* Removed debug traceback

collectors/etc/config.py

@@ -28,10 +28,6 @@
 # sys.path, so you can import other Python modules from this directory
 # or its subdirectories.
 
-from collectors.etc import yaml_conf
-
-TCOLLECTOR_CONF = yaml_conf.load_collector_configuration('xcollector.yml')['collector']
-
 def onload(options, tags):
     """Function called by tcollector when it starts up.
 
@@ -41,42 +37,42 @@ def onload(options, tags):
     """
     pass
 
+
 def get_defaults():
     """Configuration values to use as defaults in the code
 
         This is called by the OptionParser.
     """
 
-    defaults = TCOLLECTOR_CONF['config']
-    defaults['pidfile'] = "/var/run/xcollector.pid"
-    defaults['no_tcollector_stats'] = False
-    defaults['evictinterval'] = 6000
-    defaults['dedupinterval'] = 0
-    defaults['deduponlyzero'] = False
-    defaults['allowed_inactivity_time'] = 600
-    defaults['maxtags'] = 25
-    defaults['max_bytes'] = defaults['log_max_bytes']
-    defaults['http_username'] = defaults['access_token']
-    defaults['http_password'] = True
-    defaults['reconnectinterval'] = 0
-    if 'host' not in defaults:
-        defaults['host'] = "api.apptuit.ai"
-    if 'port' not in defaults:
-        defaults['port'] = 443
-    defaults['http'] = True
-    defaults['http_api_path'] = "api/put"
-    if 'tags' not in defaults:
-        defaults['tags'] = {}
-    defaults['remove_inactive_collectors'] = False
-    defaults['backup_count'] = defaults['log_backup_count']
-    defaults["cdir"] = "/usr/local/xcollector/collectors"
-    if 'ssl' not in defaults:
-        defaults['ssl'] = True
-    if 'compression_threshold' not in defaults:
-        defaults['compression_threshold'] = 6 * 1024
-    defaults['stdin'] = False
-    if 'daemonize' not in defaults:
-        defaults['daemonize'] = False
-    defaults['hosts'] = False
+    defaults = {
+        'pidfile': '/var/run/xcollector.pid',
+        'no_tcollector_stats': False,
+        'evictinterval': 6000,
+        'dedupinterval': 0,
+        'deduponlyzero': False,
+        'allowed_inactivity_time': 600,
+        'maxtags': 25,
+        'max_bytes': 64 * 1024 * 1024,
+        'http_username': False,
+        'http_password': True,
+        'reconnectinterval': 0,
+        'host': 'api.apptuit.ai',
+        'port': 443,
+        'http': True,
+        'http_api_path': "api/put",
+        'tags': [],
+        'remove_inactive_collectors': False,
+        'backup_count': 1,
+        'cdir': "/usr/local/xcollector/collectors",
+        'ydir': "/etc/xcollector/",
+        'ssl': True,
+        'compression_threshold': 6 * 1024,
+        'stdin': False,
+        'daemonize': False,
+        'hosts': False,
+        'logfile': '/var/log/xcollector/xcollector.log',
+        'verbose': False,
+        'dryrun': False,
+    }
 
     return defaults

collectors/etc/yaml_conf.py

@@ -12,7 +12,7 @@
 
 
 def get_yaml_conf_dir():
-    current_script_dir = os.path.dirname(os.path.realpath(sys.argv[0]))
+    current_script_dir = os.path.dirname(os.path.abspath(sys.argv[0]))
     if "/collectors/0" in current_script_dir:
         yaml_conf_dir = os.path.join(current_script_dir[:current_script_dir.index("/collectors/0")], 'conf')
     else:
@@ -22,6 +22,10 @@ def get_yaml_conf_dir():
 
 def load_collector_configuration(config_file):
     yaml_conf_dir = get_yaml_conf_dir()
+    return load_configuration(yaml_conf_dir, config_file)
+
+
+def load_configuration(yaml_conf_dir, config_file):
     with open(os.path.join(yaml_conf_dir, config_file), 'r') as stream:
         try:
             return yaml.safe_load(stream)

tcollector.py

@@ -73,6 +73,7 @@
 MAX_READQ_SIZE = 100000
 
 HTTP_REQUEST_TIMEOUT = 60
+YAML_CONFIG_FILE = '/etc/xcollector/xcollector.yml'
 
 
 def register_collector(collector):
@@ -592,7 +593,7 @@ def run(self):
 
                 now = int(time.time())
                 if now - send_time > 5 or len(self.sendq) > MAX_SENDQ_SIZE:
-                    LOG.info('Sender -> readerq size: %s', self.reader.readerq.qsize())
+                    LOG.debug('Sender -> readerq size: %s', self.reader.readerq.qsize())
                     self.send_data()
                     send_time = now
                 errors = 0  # We managed to do a successful iteration.
@@ -889,20 +890,6 @@ def setup_logging(logfile=DEFAULT_LOG, max_bytes=None, backup_count=None):
 def parse_cmdline(argv):
     """Parses the command-line."""
 
-    def tag_str_list_to_dict(list):
-        tag_dict = {}
-        for tag_str in list:
-            tag_value_split = tag_str.split("=")
-            if (len(tag_value_split) == 2):
-                tag_dict[tag_value_split[0].strip()] = tag_value_split[1].strip()
-        return tag_dict
-
-    def tag_dict_to_str_list(dict):
-        tag_list = []
-        for key, value in dict.items():
-            tag_list.append(key.strip() + "=" + value.strip())
-        return tag_list
-
     try:
         from collectors.etc import config
         defaults = config.get_defaults()
@@ -921,6 +908,9 @@ def tag_dict_to_str_list(dict):
     parser.add_option('-c', '--collector-dir', dest='cdir', metavar='DIR',
                       default=defaults['cdir'],
                       help=SUPPRESS_HELP)  # 'Directory where the collectors are located.'
+    parser.add_option('-y', '--yaml-conf-dir', dest='ydir', metavar='DIR',
+                      default=defaults['ydir'],
+                      help=SUPPRESS_HELP)  # 'Directory where the yaml confs are located.'
     parser.add_option('-d', '--dry-run', dest='dryrun', action='store_true',
                       default=defaults['dryrun'],
                       help=SUPPRESS_HELP)  # 'Don\'t actually send anything to the TSD, just print the datapoints.'
@@ -1010,10 +1000,11 @@ def tag_dict_to_str_list(dict):
     parser.add_option('--set-option-tags', dest='set_opt_global_tags',
                       help=SUPPRESS_HELP)  # 'Username to use for HTTP Basic Auth when sending the data via HTTP'
     (options, args) = parser.parse_args(args=argv[1:])
-    cmdline_dict = tag_str_list_to_dict(options.tags)
-    for key, value in defaults['tags'].items():
-        cmdline_dict[key] = value
-    options.tags = tag_dict_to_str_list(cmdline_dict)
+
+    global YAML_CONFIG_FILE
+    YAML_CONFIG_FILE = os.path.join(options.ydir, 'xcollector.yml')
+    load_yaml_conf(options)
+
     if options.dedupinterval < 0:
         parser.error('--dedup-interval must be at least 0 seconds')
     if options.evictinterval <= options.dedupinterval:
@@ -1027,6 +1018,20 @@ def tag_dict_to_str_list(dict):
     return (options, args)
 
 
+def load_yaml_conf(options):
+    key_mappings = {"access_token": "http_username", "log_max_bytes": "max_bytes", "log_backup_count": "backup_count"}
+    yaml_configuration = yaml_conf.load_configuration(options.ydir, 'xcollector.yml')['collector']['config']
+    for key in yaml_configuration.keys():
+        yaml_value = yaml_configuration[key]
+        if key == "tags":
+            tags = dict(item.split("=", 1) for item in options.tags)
+            for k, v in yaml_value.items():
+                tags[k] = v
+            setattr(options, key, [k + "=" + v for k, v in tags.items()])
+        else:
+            setattr(options, key_mappings.get(key, key), yaml_value)
+
+
 def daemonize():
     """Performs the necessary dance to become a background daemon."""
     if os.fork():
@@ -1102,7 +1107,7 @@ def main(argv):
 
     if not options.http_username or options.http_username == '' or options.http_username == 'PASTE_ACCESS_TOKEN_HERE':
         sys.stderr.write('access_token is not specified. Please add Apptuit issued access_token to '
-                         '/etc/xcollector/xcollector.yml\n')
+                         '%s\n' % YAML_CONFIG_FILE)
         return 3
 
     if options.daemonize: