Skip to content

Commit a994743

Browse files
santigimenosantigimeno
committed
src: enrich OTLP metrics resource attributes
Add process metadata from nsolid.info() to OTLP metric resources while leaving logs and traces on the common resource. Metrics now include different attributes derived from process `info`. This will ease our internal metrics handling. Keep resource access thread-safe by returning shared resource snapshots guarded by nsuv::ns_mutex. This avoids use-after-free if UpdateResource() replaces a resource while another thread is exporting telemetry. Update gRPC and OTLP metrics tests to validate the expanded resource attributes.
1 parent 1fec452 commit a994743

9 files changed

Lines changed: 423 additions & 61 deletions

File tree

‎agents/grpc/src/grpc_agent.cc‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -256,7 +256,8 @@ void PopulateMetricsEvent(grpcagent::MetricsEvent* metrics_event,
256256
PopulateCommon(metrics_event->mutable_common(), "metrics", req_id);
257257

258258
ResourceMetrics data;
259-
data.resource_ = otlp::GetResource();
259+
auto resource = otlp::GetMetricsResource();
260+
data.resource_ = resource.get();
260261
std::vector<MetricData> metrics;
261262

262263
// As this is the cached we're sending, we pass the same value for prev_stor.
@@ -936,7 +937,8 @@ void GrpcAgent::env_deletion_cb_(SharedEnvInst envinst,
936937
}
937938

938939
ResourceMetrics data;
939-
data.resource_ = otlp::GetResource();
940+
auto resource = otlp::GetMetricsResource();
941+
data.resource_ = resource.get();
940942
std::vector<MetricData> metrics;
941943

942944
ThreadMetricsStor stor;
@@ -1381,7 +1383,8 @@ void GrpcAgent::got_proc_metrics() {
13811383
std::vector<MetricData> metrics;
13821384
otlp::fill_proc_metrics(metrics, stor, proc_prev_stor_, false);
13831385
ResourceMetrics data;
1384-
data.resource_ = otlp::GetResource();
1386+
auto resource = otlp::GetMetricsResource();
1387+
data.resource_ = resource.get();
13851388
data.scope_metric_data_ =
13861389
std::vector<ScopeMetrics>{{otlp::GetScope(), metrics}};
13871390
auto result = metrics_exporter_->Export(data);

‎agents/otlp/src/otlp_common.cc‎

Lines changed: 231 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,17 @@
11
#include "otlp_common.h"
22
// NOLINTNEXTLINE(build/c++11)
33
#include <chrono>
4+
#include <ctime>
5+
#include <iomanip>
6+
#include <sstream>
47
#include <unordered_map>
58
#include "asserts-cpp/asserts.h"
69
#include "env-inl.h"
710
#include "nlohmann/json.hpp"
11+
#include "nsuv-inl.h"
12+
#include "opentelemetry/semconv/incubating/deployment_attributes.h"
13+
#include "opentelemetry/semconv/incubating/host_attributes.h"
14+
#include "opentelemetry/semconv/incubating/os_attributes.h"
815
#include "opentelemetry/semconv/incubating/process_attributes.h"
916
#include "opentelemetry/semconv/incubating/service_attributes.h"
1017
#include "opentelemetry/semconv/incubating/thread_attributes.h"
@@ -44,8 +51,20 @@ using opentelemetry::trace::SpanId;
4451
using opentelemetry::trace::SpanKind;
4552
using opentelemetry::trace::TraceFlags;
4653
using opentelemetry::trace::TraceId;
54+
using opentelemetry::semconv::deployment::kDeploymentEnvironmentName;
55+
using opentelemetry::semconv::host::kHostArch;
56+
using opentelemetry::semconv::host::kHostCpuModelName;
57+
using opentelemetry::semconv::host::kHostName;
58+
using opentelemetry::semconv::os::kOsType;
4759
using opentelemetry::trace::propagation::detail::HexToBinary;
60+
using opentelemetry::semconv::process::kProcessCreationTime;
61+
using opentelemetry::semconv::process::kProcessExecutablePath;
4862
using opentelemetry::semconv::process::kProcessOwner;
63+
using opentelemetry::semconv::process::kProcessPid;
64+
using opentelemetry::semconv::process::kProcessRuntimeDescription;
65+
using opentelemetry::semconv::process::kProcessRuntimeName;
66+
using opentelemetry::semconv::process::kProcessRuntimeVersion;
67+
using opentelemetry::semconv::process::kProcessTitle;
4968
using opentelemetry::semconv::service::kServiceName;
5069
using opentelemetry::semconv::service::kServiceInstanceId;
5170
using opentelemetry::semconv::service::kServiceVersion;
@@ -67,9 +86,145 @@ static std::vector<std::string> discarded_metrics = {
6786
"thread_id", "timestamp"
6887
};
6988

70-
static std::unique_ptr<Resource> resource_g =
71-
std::make_unique<Resource>(Resource::GetEmpty());
72-
static bool isResourceInitialized_g = false;
89+
static std::shared_ptr<Resource> resource_g;
90+
static std::shared_ptr<Resource> metrics_resource_g;
91+
static nsuv::ns_mutex resource_mutex_g;
92+
93+
static std::string ToIso8601(uint64_t timestamp_ms) {
94+
if (timestamp_ms == 0) return "";
95+
96+
const time_t seconds = static_cast<time_t>(timestamp_ms / 1000);
97+
std::tm tm{};
98+
#ifdef _WIN32
99+
gmtime_s(&tm, &seconds);
100+
#else
101+
gmtime_r(&seconds, &tm);
102+
#endif
103+
104+
std::ostringstream stream;
105+
stream << std::put_time(&tm, "%Y-%m-%dT%H:%M:%S");
106+
107+
const uint64_t millis = timestamp_ms % 1000;
108+
stream << '.' << std::setw(3) << std::setfill('0') << millis;
109+
110+
stream << 'Z';
111+
return stream.str();
112+
}
113+
114+
static std::string NormalizeOsType(const std::string& platform) {
115+
if (platform == "win32") return "windows";
116+
if (platform == "sunos") return "solaris";
117+
return platform;
118+
}
119+
120+
static std::string NormalizeHostArch(const std::string& arch) {
121+
if (arch == "x64") return "amd64";
122+
if (arch == "ia32") return "x86";
123+
if (arch == "arm") return "arm32";
124+
return arch;
125+
}
126+
127+
static ResourceAttributes GetMetadataResourceAttributes(const json& info) {
128+
ResourceAttributes attrs;
129+
130+
if (info.is_discarded() || !info.is_object()) return attrs;
131+
132+
auto it = info.find("app");
133+
if (it != info.end() && it->is_string()) {
134+
attrs.SetAttribute(kServiceName, it->get<std::string>());
135+
}
136+
137+
attrs.SetAttribute(kServiceInstanceId, nsolid::GetAgentId());
138+
139+
it = info.find("appVersion");
140+
if (it != info.end() && it->is_string()) {
141+
attrs.SetAttribute(kServiceVersion, it->get<std::string>());
142+
}
143+
144+
it = info.find("hostname");
145+
if (it != info.end() && it->is_string()) {
146+
attrs.SetAttribute(kHostName, it->get<std::string>());
147+
}
148+
149+
it = info.find("pid");
150+
if (it != info.end() && it->is_number_unsigned()) {
151+
attrs.SetAttribute(kProcessPid, static_cast<int64_t>(it->get<uint32_t>()));
152+
}
153+
154+
it = info.find("arch");
155+
if (it != info.end() && it->is_string()) {
156+
attrs.SetAttribute(kHostArch, NormalizeHostArch(it->get<std::string>()));
157+
}
158+
159+
it = info.find("platform");
160+
if (it != info.end() && it->is_string()) {
161+
attrs.SetAttribute(kOsType, NormalizeOsType(it->get<std::string>()));
162+
}
163+
164+
it = info.find("execPath");
165+
if (it != info.end() && it->is_string()) {
166+
attrs.SetAttribute(kProcessExecutablePath, it->get<std::string>());
167+
}
168+
169+
it = info.find("main");
170+
if (it != info.end() && it->is_string()) {
171+
attrs.SetAttribute("main", it->get<std::string>());
172+
}
173+
174+
it = info.find("nodeEnv");
175+
if (it != info.end() && it->is_string()) {
176+
attrs.SetAttribute(kDeploymentEnvironmentName, it->get<std::string>());
177+
}
178+
179+
it = info.find("versions");
180+
if (it != info.end() && it->is_object()) {
181+
auto version_it = it->find("node");
182+
if (version_it != it->end() && version_it->is_string()) {
183+
attrs.SetAttribute(kProcessRuntimeVersion,
184+
version_it->get<std::string>());
185+
}
186+
187+
version_it = it->find("nsolid");
188+
if (version_it != it->end() && version_it->is_string()) {
189+
std::string nsolid_version = version_it->get<std::string>();
190+
attrs.SetAttribute(kProcessRuntimeDescription,
191+
"N|Solid " + nsolid_version);
192+
}
193+
}
194+
195+
attrs.SetAttribute(kProcessRuntimeName, "nodejs");
196+
197+
it = info.find("cpuCores");
198+
if (it != info.end() && it->is_number_unsigned()) {
199+
attrs.SetAttribute("cpuCores", std::to_string(it->get<uint32_t>()));
200+
}
201+
202+
it = info.find("cpuModel");
203+
if (it != info.end() && it->is_string()) {
204+
attrs.SetAttribute(kHostCpuModelName, it->get<std::string>());
205+
}
206+
207+
it = info.find("processStart");
208+
if (it != info.end() && it->is_number_unsigned()) {
209+
std::string iso_time = ToIso8601(it->get<uint64_t>());
210+
if (!iso_time.empty()) {
211+
attrs.SetAttribute(kProcessCreationTime, std::move(iso_time));
212+
}
213+
}
214+
215+
it = info.find("tags");
216+
if (it != info.end() && it->is_array()) {
217+
std::string tags;
218+
for (const auto& tag : *it) {
219+
if (!tag.is_string()) continue;
220+
if (!tags.empty()) tags += ',';
221+
tags += tag.get<std::string>();
222+
}
223+
attrs.SetAttribute("tagsString", std::move(tags));
224+
}
225+
226+
return attrs;
227+
}
73228

74229
// NOLINTNEXTLINE(runtime/references)
75230
static void add_counter(std::vector<MetricData>& metrics,
@@ -142,45 +297,86 @@ InstrumentationScope* GetScope() {
142297
return scope.get();
143298
}
144299

145-
Resource* GetResource() {
146-
if (!isResourceInitialized_g) {
147-
json config = json::parse(nsolid::GetConfig(), nullptr, false);
148-
// assert because the runtime should never send me an invalid JSON config
149-
ASSERT(!config.is_discarded());
150-
auto it = config.find("app");
151-
ASSERT(it != config.end());
152-
ResourceAttributes attrs({
153-
{kServiceName, it->get<std::string>()},
154-
{kServiceInstanceId, nsolid::GetAgentId()}
155-
});
156-
157-
it = config.find("appVersion");
158-
if (it != config.end()) {
159-
attrs.SetAttribute(kServiceVersion, it->get<std::string>());
160-
}
161-
162-
// Directly construct a new Resource in the unique_ptr
163-
resource_g = std::make_unique<Resource>(Resource::Create(attrs));
164-
isResourceInitialized_g = true;
300+
static void EnsureResourceInitializedLocked() {
301+
if (resource_g != nullptr) return;
302+
303+
json config = json::parse(nsolid::GetConfig(), nullptr, false);
304+
// assert because the runtime should never send me an invalid JSON config
305+
ASSERT(!config.is_discarded());
306+
auto it = config.find("app");
307+
ASSERT(it != config.end());
308+
ResourceAttributes attrs({
309+
{kServiceName, it->get<std::string>()},
310+
{kServiceInstanceId, nsolid::GetAgentId()}
311+
});
312+
313+
it = config.find("appVersion");
314+
if (it != config.end()) {
315+
attrs.SetAttribute(kServiceVersion, it->get<std::string>());
165316
}
166317

167-
return resource_g.get();
318+
resource_g = std::make_shared<Resource>(Resource::Create(attrs));
168319
}
169320

170-
Resource* UpdateResource(ResourceAttributes&& attrs) {
321+
std::shared_ptr<Resource> GetResource() {
322+
nsuv::ns_mutex::scoped_lock lock(resource_mutex_g);
323+
EnsureResourceInitializedLocked();
324+
return resource_g;
325+
}
326+
327+
static void EnsureMetricsResourceInitializedLocked() {
328+
if (metrics_resource_g != nullptr) return;
329+
330+
EnsureResourceInitializedLocked();
331+
332+
json info = json::parse(nsolid::GetProcessInfo(), nullptr, false);
333+
ResourceAttributes attrs = GetMetadataResourceAttributes(info);
334+
auto new_res = std::make_shared<Resource>(Resource::Create(attrs));
335+
metrics_resource_g =
336+
std::make_shared<Resource>(resource_g->Merge(*new_res));
337+
}
338+
339+
std::shared_ptr<Resource> GetMetricsResource() {
340+
nsuv::ns_mutex::scoped_lock lock(resource_mutex_g);
341+
EnsureMetricsResourceInitializedLocked();
342+
return metrics_resource_g;
343+
}
344+
345+
std::shared_ptr<Resource> UpdateResource(ResourceAttributes&& attrs) {
346+
nsuv::ns_mutex::scoped_lock lock(resource_mutex_g);
347+
EnsureResourceInitializedLocked();
348+
349+
ResourceAttributes metrics_attrs(attrs);
350+
171351
// First, get current kServiceName to avoid overwriting it with the default
172352
// value "unknown_service". (See Resource::Create() method in the SDK).
173-
auto resource = GetResource();
174-
auto attributes = resource->GetAttributes();
353+
auto attributes = resource_g->GetAttributes();
175354
if (attributes.find(kServiceName) != attributes.end() &&
176355
attrs.find(kServiceName) == attrs.end()) {
177356
attrs.SetAttribute(kServiceName,
178357
opentelemetry::nostd::get<std::string>(attributes[kServiceName]));
179358
}
180359

181-
auto new_res = std::make_unique<Resource>(Resource::Create(attrs));
182-
resource_g = std::make_unique<Resource>(resource->Merge(*new_res));
183-
return resource_g.get();
360+
auto new_res = std::make_shared<Resource>(Resource::Create(attrs));
361+
resource_g = std::make_shared<Resource>(resource_g->Merge(*new_res));
362+
363+
if (metrics_resource_g != nullptr) {
364+
auto metrics_attributes = metrics_resource_g->GetAttributes();
365+
if (metrics_attributes.find(kServiceName) != metrics_attributes.end() &&
366+
metrics_attrs.find(kServiceName) == metrics_attrs.end()) {
367+
metrics_attrs.SetAttribute(
368+
kServiceName,
369+
opentelemetry::nostd::get<std::string>(
370+
metrics_attributes[kServiceName]));
371+
}
372+
373+
auto metrics_new_res =
374+
std::make_shared<Resource>(Resource::Create(metrics_attrs));
375+
metrics_resource_g =
376+
std::make_shared<Resource>(metrics_resource_g->Merge(*metrics_new_res));
377+
}
378+
379+
return resource_g;
184380
}
185381

186382
// NOLINTNEXTLINE(runtime/references)
@@ -250,7 +446,7 @@ NSOLID_PROCESS_METRICS_DOUBLE(V)
250446
if (prev_stor.user != stor.user || prev_stor.title != stor.title) {
251447
ResourceAttributes attrs = {
252448
{ kProcessOwner, stor.user },
253-
{ "process.title", stor.title },
449+
{ kProcessTitle, stor.title },
254450
};
255451

256452
USE(UpdateResource(std::move(attrs)));
@@ -370,7 +566,8 @@ void fill_log_recordable(LogsRecordable* recordable,
370566
nanoseconds(static_cast<uint64_t>(info.timestamp))));
371567
recordable->SetTimestamp(ts);
372568
recordable->SetObservedTimestamp(ts);
373-
recordable->SetResource(*GetResource());
569+
auto resource = GetResource();
570+
recordable->SetResource(*resource);
374571
recordable->SetInstrumentationScope(*GetScope());
375572
}
376573

@@ -508,7 +705,8 @@ void fill_recordable(Recordable* recordable, const Tracer::SpanStor& s) {
508705
recordable->SetAttribute("thread.id", s.thread_id);
509706
recordable->SetAttribute("nsolid.span_type", s.type);
510707

511-
recordable->SetResource(*GetResource());
708+
auto resource = GetResource();
709+
recordable->SetResource(*resource);
512710
}
513711

514712
} // namespace otlp

‎agents/otlp/src/otlp_common.h‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
#ifndef AGENTS_OTLP_SRC_OTLP_COMMON_H_
22
#define AGENTS_OTLP_SRC_OTLP_COMMON_H_
33

4+
#include <memory>
5+
46
#include "nsolid.h"
57
#include "opentelemetry/sdk/metrics/data/metric_data.h"
68
#include "opentelemetry/sdk/resource/resource.h"
@@ -42,9 +44,14 @@ namespace otlp {
4244
OPENTELEMETRY_NAMESPACE::sdk::instrumentationscope::InstrumentationScope*
4345
GetScope();
4446

45-
OPENTELEMETRY_NAMESPACE::sdk::resource::Resource* GetResource();
47+
std::shared_ptr<OPENTELEMETRY_NAMESPACE::sdk::resource::Resource>
48+
GetResource();
49+
50+
std::shared_ptr<OPENTELEMETRY_NAMESPACE::sdk::resource::Resource>
51+
GetMetricsResource();
4652

47-
OPENTELEMETRY_NAMESPACE::sdk::resource::Resource* UpdateResource(
53+
std::shared_ptr<OPENTELEMETRY_NAMESPACE::sdk::resource::Resource>
54+
UpdateResource(
4855
OPENTELEMETRY_NAMESPACE::sdk::resource::ResourceAttributes&&);
4956

5057
void fill_proc_metrics(std::vector<opentelemetry::sdk::metrics::MetricData>&,

0 commit comments

Comments
 (0)