Change pid to process_token

The "pid" value that is collected by the client plugin and passed to the
aggregator in order to disambiguate separate processes is now augmented
by a six character random hex string, so that a single host that may have
the same pid repeated, such as when process namespaces or containers are
used, sends correct statistics for the same program name configuration.

Fixes: #9
Change-Id: I02081ef0e27703e27dc0bceb63566acaf608f777

Author: zzzeek

sqlalchemy_collectd/client/sender.py

@@ -38,12 +38,12 @@ def __init__(
             [protocol_type for protocol_type, sender in senders],
         )
 
-    def send(self, collection_target, timestamp, interval, pid):
+    def send(self, collection_target, timestamp, interval, process_token):
         values = protocol.Values(
             host=self.hostname,
             plugin=self.plugin,
             plugin_instance=self.stats_name,
-            type_instance=str(pid),
+            type_instance=str(process_token),
             interval=interval,
             time=timestamp,
         )

sqlalchemy_collectd/client/tests/test_worker.py

@@ -75,7 +75,11 @@ def start_loop():
 
         self.assertEqual(
             [
-                mock.call.info("Starting process thread in pid %s", mock.ANY),
+                mock.call.info(
+                    "Starting process thread in pid %s, process token %s",
+                    mock.ANY,
+                    mock.ANY,
+                ),
                 mock.call.error("error sending stats", exc_info=True),
                 mock.call.info(
                     "message sender thread caught SystemExit "

sqlalchemy_collectd/client/worker.py

@@ -2,6 +2,7 @@
 import os
 import threading
 import time
+import uuid
 
 log = logging.getLogger(__name__)
 
@@ -25,7 +26,12 @@ def _check_threads_started():
 
 def _process(interval):
     pid = os.getpid()
-    log.info("Starting process thread in pid %s", pid)
+    process_token = "%s:%s" % (pid, str(uuid.uuid4())[0:6])
+    log.info(
+        "Starting process thread in pid %s, process token %s",
+        pid,
+        process_token,
+    )
 
     try:
         while True:
@@ -37,7 +43,9 @@ def _process(interval):
                 if now - last_called[0] > interval:
                     last_called[0] = now
                     try:
-                        sender.send(collection_target, now, interval, pid)
+                        sender.send(
+                            collection_target, now, interval, process_token
+                        )
                     except Exception:
                         log.error("error sending stats", exc_info=True)
 

sqlalchemy_collectd/server/receiver.py

@@ -57,23 +57,26 @@ def _set_stats(self, values):
         timestamp = values.time
         hostname = values.host
         progname = values.plugin_instance
-        pid = values.type_instance
+        process_token = values.type_instance
         interval = values.interval
 
         bucket = self.buckets[bucket_name]
         records = bucket.get_data(timestamp, interval=interval * 2)
 
-        records[(hostname, progname, pid)] = values
+        records[(hostname, progname, process_token)] = values
 
-        if pid:
+        if process_token:
             # manufacture a record for that is a single process count for this
-            # pid. we also use a larger interval
+            # process_token (which is roughly the pid plus a unique key
+            # generated by the client plugin).   we also use a larger interval
             # for this value so that the process count changes more slowly
             process_bucket = self.buckets[internal_types.process_internal.name]
             process_records = process_bucket.get_data(
                 timestamp, interval=interval * 5
             )
-            process_records[(hostname, progname, pid)] = values.build(
+            process_records[
+                (hostname, progname, process_token)
+            ] = values.build(
                 type=internal_types.process_internal.name, values=[1]
             )
 
@@ -86,8 +89,8 @@ def get_stats_by_progname(self, bucket_name, timestamp, agg_func=sum):
         ):
             recs = [records[key] for key in keys]
             interval = recs[0].interval
-            # summation here is across pids.
-            # if records are pid-less, then there would be one record
+            # summation here is across process_tokens.
+            # if records are process_token-less, then there would be one record
             # per host/program name.
             values_obj = agg_func(recs).build(
                 time=timestamp, interval=interval

unreleased_changes/9.rst

@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug
+    :tickets: 9
+
+    The "pid" value that is collected by the client plugin and passed to the
+    aggregator in order to disambiguate separate processes is now augmented
+    by a six character random hex string, so that a single host that may have
+    the same pid repeated, such as when process namespaces or containers are
+    used, sends correct statistics for the same program name configuration.
+