nsolid/agents/otlp/src/otlp_agent.h at 323584527bb5e5e132ed312e05b4b9427d64e3fe · nodesource/nsolid · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
#ifndef AGENTS_OTLP_SRC_OTLP_AGENT_H_
#define AGENTS_OTLP_SRC_OTLP_AGENT_H_

#include <nsolid.h>
#include <nsolid/thread_safe.h>
#include <memory>
#include "nlohmann/json.hpp"

#include "asserts-cpp/asserts.h"
#include "metrics_exporter.h"
#include "opentelemetry/version.h"


// Class pre-declaration
OPENTELEMETRY_BEGIN_NAMESPACE
namespace exporter {
namespace otlp {
struct OtlpHttpExporterOptions;
struct OtlpGrpcExporterOptions;
}
}
namespace sdk {
namespace trace {
class Recordable;
class SpanExporter;
}
}
OPENTELEMETRY_END_NAMESPACE

using UniqRecordable =
    std::unique_ptr<OPENTELEMETRY_NAMESPACE::sdk::trace::Recordable>;
using UniqRecordables = std::vector<UniqRecordable>;

namespace node {
namespace nsolid {

class SpanCollector;

namespace otlp {

struct JSThreadMetrics {
  explicit JSThreadMetrics(SharedEnvInst envinst);
  SharedThreadMetrics metrics_;
  ThreadMetrics::MetricsStor prev_;
  double loop_start_;
};

class OTLPAgent {
 public:
  static OTLPAgent* Inst();

  int start();

  int stop();

  int config(const nlohmann::json& config);

 private:
  friend class OTLPMetrics;
  OTLPAgent();

  ~OTLPAgent();

  static void run_(nsuv::ns_thread*, OTLPAgent*);

  static void shutdown_cb_(nsuv::ns_async*, OTLPAgent*);

  static void config_agent_cb_(std::string, OTLPAgent*);

  static void config_msg_cb_(nsuv::ns_async*, OTLPAgent*);

  static void env_msg_cb_(nsuv::ns_async*, OTLPAgent*);

  static void on_thread_add_(SharedEnvInst, OTLPAgent* agent);

  static void on_thread_remove_(SharedEnvInst, OTLPAgent* agent);

  static void metrics_timer_cb_(nsuv::ns_timer*, OTLPAgent*);

  static void metrics_msg_cb_(nsuv::ns_async*, OTLPAgent* agent);

  static void thr_metrics_cb_(SharedThreadMetrics, OTLPAgent*);

  static UniqRecordable transf(const Tracer::SpanStor& span, OTLPAgent* agent);

  void do_start();

  void do_stop();

  void config_datadog(const nlohmann::json& config);

  void config_dynatrace(const nlohmann::json& config);

  void config_endpoint(const std::string& type,
                       const nlohmann::json& config);

  void config_newrelic(const nlohmann::json& config);

  void config_otlp_agent(const nlohmann::json& config);

  void config_otlp_endpoint(const nlohmann::json& config);

  void got_proc_metrics();

  void got_spans(const UniqRecordables& spans);

  void setup_trace_otlp_exporter(  // NOLINTNEXTLINE(runtime/references)
    OPENTELEMETRY_NAMESPACE::exporter::otlp::OtlpHttpExporterOptions& opts);

  void setup_trace_grpc_otlp_exporter(  // NOLINTNEXTLINE(runtime/references)
    OPENTELEMETRY_NAMESPACE::exporter::otlp::OtlpGrpcExporterOptions& opts);

  int setup_metrics_timer(uint64_t period);

  void update_tracer(uint32_t flags);

  uv_loop_t loop_;
  nsuv::ns_thread thread_;
  nsuv::ns_async shutdown_;

  // For otlp thread start synchronization
  std::atomic<bool> ready_;
  uv_cond_t start_cond_;
  uv_mutex_t start_lock_;

  bool hooks_init_;

  nsuv::ns_async env_msg_;
  TSQueue<std::tuple<SharedEnvInst, bool>> env_msg_q_;

  // For the Tracing API
  uint32_t trace_flags_;
  std::shared_ptr<SpanCollector> span_collector_;

  std::unique_ptr<OPENTELEMETRY_NAMESPACE::sdk::trace::SpanExporter>
    otlp_exporter_;

  // For the Metrics API
  uint64_t metrics_interval_;
  ProcessMetrics proc_metrics_;
  ProcessMetrics::MetricsStor proc_prev_stor_;
  std::map<uint64_t, JSThreadMetrics> env_metrics_map_;
  nsuv::ns_async metrics_msg_;
  TSQueue<ThreadMetrics::MetricsStor> thr_metrics_msg_q_;
  nsuv::ns_timer metrics_timer_;
  std::unique_ptr<MetricsExporter> metrics_exporter_;

  // For the Configuration API
  nsuv::ns_async config_msg_;
  TSQueue<nlohmann::json> config_msg_q_;
  nlohmann::json config_;

  std::string cacert_;
};

}  // namespace otlp
}  // namespace nsolid
}  // namespace node

#endif  // AGENTS_OTLP_SRC_OTLP_AGENT_H_