Sanchda/add metrics

CMakeLists.txt

@@ -192,7 +192,6 @@ endif()
 
 set(DD_PROFILING_SOURCES
     # cmake-format: sortable
-    src/daemonize.cc
     src/ddprof_cmdline.cc
     src/ddres_list.cc
     src/ipc.cc

include/atomic_shared.hpp

@@ -0,0 +1,57 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0. This product includes software
+// developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present
+// Datadog, Inc.
+
+#pragma once
+#include <atomic>
+#include <chrono>
+#include <new>
+#include <thread>
+
+#include <sys/mman.h>
+
+template <class T> class AtomicShared : public std::atomic<T> {
+public:
+  static void *operator new(size_t) {
+    void *const pv = mmap(0, sizeof(T), PROT_READ | PROT_WRITE,
+                          MAP_SHARED | MAP_ANONYMOUS, -1, 0);
+    if (pv == MAP_FAILED)
+      throw std::bad_alloc();
+    return pv;
+  };
+  static void operator delete(void *pv) { munmap(pv, sizeof(T)); };
+  AtomicShared &operator=(const int b) {
+    std::atomic<pid_t>::operator=(b);
+    return *this;
+  }
+
+  bool value_timedwait(const T &oldval, int timeout) {
+    // Block until the value is different from oldval.  If the timeout is 0,
+    // check once without blocking.  If the value is negative, then block
+    // indefinitely (may be "expensive" in some sense).
+    // Doesn't do anything fancy to enforce re-scheduling the thread when
+    // the condition occurs, nor to decrease sleep overhead.  As per the spec,
+    // doesn't protect against the ABA problem (A changes to B, then back to A,
+    // before B can be detected in the loop).
+    // Will perform three "fast checks" before starting to yield to the
+    // scheduler.  This will appear as a hotspot when the caller has to wait
+    // a lot.
+    auto start = std::chrono::high_resolution_clock::now();
+    auto end = std::chrono::milliseconds(timeout);
+    if (timeout < 0) {
+      end = std::chrono::hours::max();
+    }
+    int fast_checks = 3; // hardcoded
+    do {
+      if (this->load() != oldval) {
+        return true;
+      } else if (fast_checks > 0) {
+        --fast_checks;
+      } else {
+        std::this_thread::yield();
+      }
+    } while (std::chrono::high_resolution_clock::now() - start < end);
+    return false;
+  }
+};

include/daemonize.hpp

@@ -5,20 +5,108 @@
 
 #pragma once
 
+#include "atomic_shared.hpp"
+
 #include <functional>
 #include <sys/types.h>
+#include <unistd.h>
+
+#include "daemonize.hpp"
+#include "ddprof_exit.hpp"
+#include "logger.hpp"
 
 namespace ddprof {
 
+enum class DaemonizeState { Failure = -1, Invoker = 0, Daemon = 1 };
+
 struct DaemonizeResult {
-  pid_t
-      temp_pid; // -1 on failure, 0 for initial process, > 0 for daemon process
-  pid_t parent_pid; // pid of process initiating daemonize
-  pid_t daemon_pid; // pid of daemon process
-};
+  DaemonizeState state = DaemonizeState::Failure;
+  pid_t invoker_pid = -1;
+  pid_t daemon_pid = -1;
+
+  // A reusable barrier between parent/daemon
+  void barrier() {
+    switch (state) {
+    case DaemonizeState::Failure:
+      break;
+    case DaemonizeState::Daemon:
+      sem_daemon->store(true);
+      sem_invoker->value_timedwait(false, timeout_val);
+      sem_invoker->store(false);
+      break;
+    case DaemonizeState::Invoker:
+      sem_invoker->store(true);
+      sem_daemon->value_timedwait(false, timeout_val);
+      sem_daemon->store(false);
+      break;
+    }
+  };
+
+  bool is_failure() { return state == DaemonizeState::Failure; }
+  bool is_daemon() { return state == DaemonizeState::Daemon; }
+  bool is_invoker() { return state == DaemonizeState::Invoker; }
+
+  // Daemonization function
+  // cleanup_function is a callable invoked in the context of the intermediate,
+  // short-lived process that will be killed by daemon process.
+  bool daemonize(std::function<void()> cleanup_function = {}) {
+    // Initialize
+    pid_transfer = std::make_unique<AtomicShared<pid_t>>();
+    sem_invoker = std::make_unique<AtomicShared<bool>>();
+    sem_daemon = std::make_unique<AtomicShared<bool>>();
+    pid_transfer->store(0);
+    sem_invoker->store(false);
+    sem_daemon->store(false);
 
-// Daemonization function
-// cleanup_function is a callable invoked in the context of the intermediate,
-// short-lived process that will be killed by daemon process.
-DaemonizeResult daemonize(std::function<void()> cleanup_function = {});
-} // namespace ddprof
+    // Start daemonizing
+    invoker_pid = getpid();
+    pid_t temp_pid = fork(); // "middle"/"child" (temporary) PID
+    state = DaemonizeState::Failure;
+
+    if (!temp_pid) { // temp PID enter branch
+      if (fork()) {  // temp PID enter branch
+        if (cleanup_function) {
+          cleanup_function();
+        }
+
+        // Temporary PID exits to force re-init of grandchild (daemon)
+        throw ddprof::exit();
+        std::exit(0);
+
+      } else { // grandchild (daemon) PID enter branch
+        daemon_pid = getpid();
+
+        // Tell the invoker my PID, then wait until it gets changed again.
+        pid_transfer->store(daemon_pid);
+        if (!pid_transfer->value_timedwait(daemon_pid, timeout_val)) {
+          return false;
+        }
+        state = DaemonizeState::Daemon;
+        return true;
+      }
+    } else if (temp_pid != -1) { // parent PID enter branch
+      // Try to read the PID of the daemon from shared memory, then notify
+      if (!pid_transfer->value_timedwait(0, timeout_val)) {
+        return false;
+      }
+      daemon_pid = pid_transfer->exchange(0); // consume + reset
+      state = DaemonizeState::Invoker;
+      return true;
+    }
+
+    // Should only arrive here if the first-level fork failed, but add sink to
+    if (getpid() != invoker_pid) {
+      LG_WRN("Extraneous PID (%d) detected", getpid());
+      throw ddprof::exit();
+      std::exit(-1);
+    }
+    return false;
+  }
+
+private:
+  std::unique_ptr<AtomicShared<pid_t>> pid_transfer;
+  std::unique_ptr<AtomicShared<bool>> sem_invoker;
+  std::unique_ptr<AtomicShared<bool>> sem_daemon;
+  static constexpr size_t timeout_val = 1000; // 10 seconds
+};
+} // namespace ddprof

include/ddprof_cmdline.hpp

@@ -8,6 +8,10 @@
 #include <stddef.h> // size_t
 #include <stdint.h> // uint64_t
 
+#include <string>
+#include <string_view>
+#include <vector>
+
 typedef struct PerfWatcher PerfWatcher;
 
 /**************************** Cmdline Helpers *********************************/
@@ -22,9 +26,13 @@ typedef struct PerfWatcher PerfWatcher;
 
 /// Returns index to element that compars to str, otherwise -1
 int arg_which(const char *str, char const *const *set, int sz_set);
+int arg_which(const char *str, const std::vector<std::string_view> &list);
 
 bool arg_inset(const char *str, char const *const *set, int sz_set);
 
-bool arg_yesno(const char *str, int mode);
+bool is_yes(const char *str);
+bool is_yes(const std::string &str);
+bool is_no(const char *str);
+bool is_no(const std::string &str);
 
 bool watcher_from_str(const char *str, PerfWatcher *watcher);

include/ddprof_context.hpp

@@ -10,9 +10,12 @@
 #include "ddprof_defs.hpp"
 #include "ddprof_worker_context.hpp"
 #include "exporter_input.hpp"
+#include "logger.hpp"
+#include "metric_aggregator.hpp"
 #include "perf_watcher.hpp"
 
 #include <sched.h>
+#include <string>
 
 // forward declarations
 typedef struct StackHandler StackHandler;
@@ -33,9 +36,9 @@ typedef struct DDProfContext {
     bool wait_on_socket;
     bool show_samples;
     cpu_set_t cpu_affinity;
-    const char *switch_user;
-    const char *internal_stats;
-    const char *tags;
+    std::string switch_user;
+    std::string internal_stats;
+    std::string tags;
   } params;
 
   bool initialized;
@@ -45,4 +48,7 @@ typedef struct DDProfContext {
   int num_watchers;
   void *callback_ctx; // user state to be used in callback (lib mode)
   DDProfWorkerContext worker_ctx;
+  MetricAggregator metrics;
+
+  void release() noexcept;
 } DDProfContext;

include/ddprof_exit.hpp

@@ -0,0 +1,10 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0. This product includes software
+// developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present
+// Datadog, Inc.
+
+#pragma once
+
+namespace ddprof {
+struct exit : public std::exception {};
+} // namespace ddprof

include/ddprof_input.hpp

@@ -5,6 +5,8 @@
 
 #pragma once
 
+#include <string>
+
 #include "ddprof_defs.hpp"
 #include "ddres_def.hpp"
 #include "exporter_input.hpp"
@@ -16,28 +18,29 @@ typedef int16_t watcher_index_t;
 typedef struct DDProfInput {
   int nb_parsed_params;
   // Parameters for interpretation
-  char *log_mode;
-  char *log_level;
+  std::string log_mode;
+  std::string log_level;
   // Input parameters
-  char *show_config;
-  char *show_samples;
-  char *affinity;
-  char *enable;
-  char *native_enable;
-  char *agentless;
-  char *upload_period;
-  char *fault_info;
-  char *core_dumps;
-  char *nice;
-  char *pid;
-  char *global;
-  char *worker_period;
-  char *internal_stats;
-  char *tags;
-  char *url;
-  char *socket;
-  char *preset;
-  char *switch_user;
+  std::string show_config;
+  std::string show_samples;
+  std::string affinity;
+  std::string enable;
+  std::string native_enable;
+  std::string agentless;
+  std::string upload_period;
+  std::string fault_info;
+  std::string core_dumps;
+  std::string nice;
+  std::string pid;
+  std::string global;
+  std::string worker_period;
+  std::string internal_stats;
+  std::string tags;
+  std::string url;
+  std::string socket;
+  std::string metrics_socket;
+  std::string preset;
+  std::string switch_user;
   // Watcher presets
   PerfWatcher watchers[MAX_TYPE_WATCHER];
   int num_watchers;
@@ -71,38 +74,39 @@ typedef struct DDProfInput {
 */
 // clang-format off
 
-//  A                                   B                   C   D   E   F     G     H             I
-#define OPT_TABLE(XX)                                                                                          \
-  XX(DD_API_KEY,                        api_key,            A, 'A', 1, input, NULL, "",           exp_input.)  \
-  XX(DD_ENV,                            environment,        E, 'E', 1, input, NULL, "",           exp_input.)  \
-  XX(DD_AGENT_HOST,                     host,               H, 'H', 1, input, NULL, "localhost",  exp_input.)  \
-  XX(DD_SITE,                           site,               I, 'I', 1, input, NULL, "",           exp_input.)  \
-  XX(DD_TRACE_AGENT_PORT,               port,               P, 'P', 1, input, NULL, "8126",       exp_input.)  \
-  XX(DD_TRACE_AGENT_URL,                url,                U, 'U', 1, input, NULL, "", )                      \
-  XX(DD_SERVICE,                        service,            S, 'S', 1, input, NULL, "myservice",  exp_input.)  \
-  XX(DD_VERSION,                        service_version,    V, 'V', 1, input, NULL, "",           exp_input.)  \
-  XX(DD_PROFILING_EXPORT,               do_export,          X, 'X', 1, input, NULL, "yes",        exp_input.)  \
-  XX(DD_PROFILING_PPROF_PREFIX,         debug_pprof_prefix, O, 'O', 1, input, NULL, "",           exp_input.)  \
-  XX(DD_PROFILING_AGENTLESS,            agentless,          L, 'L', 1, input, NULL, "", )                      \
-  XX(DD_TAGS,                           tags,               T, 'T', 1, input, NULL, "", )                      \
-  XX(DD_PROFILING_ENABLED,              enable,             d, 'd', 1, input, NULL, "yes", )                   \
-  XX(DD_PROFILING_NATIVE_ENABLED,       native_enable,      n, 'n', 1, input, NULL, "", )                      \
-  XX(DD_PROFILING_UPLOAD_PERIOD,        upload_period,      u, 'u', 1, input, NULL, "59", )                    \
-  XX(DD_PROFILING_NATIVE_WORKER_PERIOD, worker_period,      w, 'w', 1, input, NULL, "240", )                   \
-  XX(DD_PROFILING_NATIVE_FAULT_INFO,    fault_info,         s, 's', 1, input, NULL, "yes", )                   \
-  XX(DD_PROFILING_NATIVE_CORE_DUMPS,    core_dumps,         m, 'm', 1, input, NULL, "no", )                    \
-  XX(DD_PROFILING_NATIVE_NICE,          nice,               i, 'i', 1, input, NULL, "", )                      \
-  XX(DD_PROFILING_NATIVE_SHOW_CONFIG,   show_config,        c, 'c', 1, input, NULL, "no", )                    \
-  XX(DD_PROFILING_NATIVE_LOG_MODE,      log_mode,           o, 'o', 1, input, NULL, "stdout", )                \
-  XX(DD_PROFILING_NATIVE_LOG_LEVEL,     log_level,          l, 'l', 1, input, NULL, "error", )                 \
-  XX(DD_PROFILING_NATIVE_TARGET_PID,    pid,                p, 'p', 1, input, NULL, "", )                      \
-  XX(DD_PROFILING_NATIVE_GLOBAL,        global,             g, 'g', 1, input, NULL, "", )                      \
-  XX(DD_PROFILING_INTERNAL_STATS,       internal_stats,     b, 'b', 1, input, NULL, "", )                      \
-  XX(DD_PROFILING_NATIVE_SOCKET,        socket,             z, 'z', 1, input, NULL, "", )                      \
-  XX(DD_PROFILING_NATIVE_PRESET,        preset,             D, 'D', 1, input, NULL, "", )                      \
-  XX(DD_PROFILING_NATIVE_SHOW_SAMPLES,  show_samples,       y, 'y', 0, input, NULL, "", )                      \
-  XX(DD_PROFILING_NATIVE_CPU_AFFINITY,  affinity,           a, 'a', 1, input, NULL, "", )                      \
-  XX(DD_PROFILING_NATIVE_SWITCH_USER,   switch_user,        W, 'W', 1, input, NULL, "", )
+//  A                                    B                   C   D   E   F     G     H             I
+#define OPT_TABLE(XX)                                                                                                         \
+  XX(DD_API_KEY,                         api_key,            A, 'A', 1, input, NULL, "",           exp_input.)                \
+  XX(DD_ENV,                             environment,        E, 'E', 1, input, NULL, "",           exp_input.)                \
+  XX(DD_AGENT_HOST,                      host,               H, 'H', 1, input, NULL, "localhost",  exp_input.)                \
+  XX(DD_SITE,                            site,               I, 'I', 1, input, NULL, "",           exp_input.)                \
+  XX(DD_TRACE_AGENT_PORT,                port,               P, 'P', 1, input, NULL, "8126",       exp_input.)                \
+  XX(DD_TRACE_AGENT_URL,                 url,                U, 'U', 1, input, NULL, "", )                                    \
+  XX(DD_SERVICE,                         service,            S, 'S', 1, input, NULL, "myservice",  exp_input.)                \
+  XX(DD_VERSION,                         service_version,    V, 'V', 1, input, NULL, "",           exp_input.)                \
+  XX(DD_PROFILING_EXPORT,                do_export,          X, 'X', 1, input, NULL, "yes",        exp_input.)                \
+  XX(DD_PROFILING_PPROF_PREFIX,          debug_pprof_prefix, O, 'O', 1, input, NULL, "",           exp_input.)                \
+  XX(DD_PROFILING_AGENTLESS,             agentless,          L, 'L', 1, input, NULL, "", )                                    \
+  XX(DD_TAGS,                            tags,               T, 'T', 1, input, NULL, "", )                                    \
+  XX(DD_PROFILING_ENABLED,               enable,             d, 'd', 1, input, NULL, "yes", )                                 \
+  XX(DD_PROFILING_NATIVE_ENABLED,        native_enable,      n, 'n', 1, input, NULL, "", )                                    \
+  XX(DD_PROFILING_UPLOAD_PERIOD,         upload_period,      u, 'u', 1, input, NULL, "59", )                                  \
+  XX(DD_PROFILING_NATIVE_WORKER_PERIOD,  worker_period,      w, 'w', 1, input, NULL, "240", )                                 \
+  XX(DD_PROFILING_NATIVE_FAULT_INFO,     fault_info,         s, 's', 1, input, NULL, "yes", )                                 \
+  XX(DD_PROFILING_NATIVE_CORE_DUMPS,     core_dumps,         m, 'm', 1, input, NULL, "no", )                                  \
+  XX(DD_PROFILING_NATIVE_NICE,           nice,               i, 'i', 1, input, NULL, "", )                                    \
+  XX(DD_PROFILING_NATIVE_SHOW_CONFIG,    show_config,        c, 'c', 1, input, NULL, "no", )                                  \
+  XX(DD_PROFILING_NATIVE_LOG_MODE,       log_mode,           o, 'o', 1, input, NULL, "stdout", )                              \
+  XX(DD_PROFILING_NATIVE_LOG_LEVEL,      log_level,          l, 'l', 1, input, NULL, "error", )                               \
+  XX(DD_PROFILING_NATIVE_TARGET_PID,     pid,                p, 'p', 1, input, NULL, "", )                                    \
+  XX(DD_PROFILING_NATIVE_GLOBAL,         global,             g, 'g', 1, input, NULL, "", )                                    \
+  XX(DD_PROFILING_INTERNAL_STATS,        internal_stats,     b, 'b', 1, input, NULL, "", )                                    \
+  XX(DD_PROFILING_NATIVE_SOCKET,         socket,             z, 'z', 1, input, NULL, "", )                                    \
+  XX(DD_PROFILING_NATIVE_METRICS_SOCKET, metrics_socket,     k, 'k', 1, input, NULL, "/var/run/datadog-agent/statsd.sock", )  \
+  XX(DD_PROFILING_NATIVE_PRESET,         preset,             D, 'D', 1, input, NULL, "", )                                    \
+  XX(DD_PROFILING_NATIVE_SHOW_SAMPLES,   show_samples,       y, 'y', 0, input, NULL, "", )                                    \
+  XX(DD_PROFILING_NATIVE_CPU_AFFINITY,   affinity,           a, 'a', 1, input, NULL, "", )                                    \
+  XX(DD_PROFILING_NATIVE_SWITCH_USER,    switch_user,        W, 'W', 1, input, NULL, "", )
 // clang-format on
 
 #define X_ENUM(a, b, c, d, e, f, g, h, i) a,
@@ -122,5 +126,3 @@ void ddprof_print_help();
 
 // Print help
 void ddprof_print_params(const DDProfInput *input);
-
-void ddprof_input_free(DDProfInput *input);