Skip to content

Commit 26e038d

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 f397c0b commit 26e038d

9 files changed

Lines changed: 435 additions & 81 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: 248 additions & 53 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,230 @@ 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+
92+
static nsuv::ns_mutex& ResourceMutex() {
93+
static int er = 0;
94+
static nsuv::ns_mutex mutex(&er, false);
95+
ASSERT_EQ(0, er);
96+
return mutex;
97+
}
98+
99+
static std::string ToIso8601(uint64_t timestamp_ms) {
100+
if (timestamp_ms == 0) return "";
101+
102+
const time_t seconds = static_cast<time_t>(timestamp_ms / 1000);
103+
std::tm tm{};
104+
#ifdef _WIN32
105+
gmtime_s(&tm, &seconds);
106+
#else
107+
gmtime_r(&seconds, &tm);
108+
#endif
109+
110+
std::ostringstream stream;
111+
stream << std::put_time(&tm, "%Y-%m-%dT%H:%M:%S");
112+
113+
const uint64_t millis = timestamp_ms % 1000;
114+
stream << '.' << std::setw(3) << std::setfill('0') << millis;
115+
116+
stream << 'Z';
117+
return stream.str();
118+
}
119+
120+
static std::string NormalizeOsType(const std::string& platform) {
121+
if (platform == "win32") return "windows";
122+
if (platform == "sunos") return "solaris";
123+
return platform;
124+
}
125+
126+
static std::string NormalizeHostArch(const std::string& arch) {
127+
if (arch == "x64") return "amd64";
128+
if (arch == "ia32") return "x86";
129+
if (arch == "arm") return "arm32";
130+
return arch;
131+
}
132+
133+
static ResourceAttributes GetMetadataResourceAttributes(const json& info) {
134+
ResourceAttributes attrs;
135+
136+
if (info.is_discarded() || !info.is_object()) return attrs;
137+
138+
auto it = info.find("app");
139+
if (it != info.end() && it->is_string()) {
140+
attrs.SetAttribute(kServiceName, it->get<std::string>());
141+
}
142+
143+
attrs.SetAttribute(kServiceInstanceId, nsolid::GetAgentId());
144+
145+
it = info.find("appVersion");
146+
if (it != info.end() && it->is_string()) {
147+
attrs.SetAttribute(kServiceVersion, it->get<std::string>());
148+
}
149+
150+
it = info.find("hostname");
151+
if (it != info.end() && it->is_string()) {
152+
attrs.SetAttribute(kHostName, it->get<std::string>());
153+
}
154+
155+
it = info.find("pid");
156+
if (it != info.end() && it->is_number_unsigned()) {
157+
attrs.SetAttribute(kProcessPid, static_cast<int64_t>(it->get<uint32_t>()));
158+
}
159+
160+
it = info.find("arch");
161+
if (it != info.end() && it->is_string()) {
162+
attrs.SetAttribute(kHostArch, NormalizeHostArch(it->get<std::string>()));
163+
}
164+
165+
it = info.find("platform");
166+
if (it != info.end() && it->is_string()) {
167+
attrs.SetAttribute(kOsType, NormalizeOsType(it->get<std::string>()));
168+
}
169+
170+
it = info.find("execPath");
171+
if (it != info.end() && it->is_string()) {
172+
attrs.SetAttribute(kProcessExecutablePath, it->get<std::string>());
173+
}
174+
175+
it = info.find("main");
176+
if (it != info.end() && it->is_string()) {
177+
attrs.SetAttribute("main", it->get<std::string>());
178+
}
179+
180+
it = info.find("nodeEnv");
181+
if (it != info.end() && it->is_string()) {
182+
attrs.SetAttribute(kDeploymentEnvironmentName, it->get<std::string>());
183+
}
184+
185+
it = info.find("versions");
186+
if (it != info.end() && it->is_object()) {
187+
auto version_it = it->find("node");
188+
if (version_it != it->end() && version_it->is_string()) {
189+
attrs.SetAttribute(kProcessRuntimeVersion,
190+
version_it->get<std::string>());
191+
}
192+
193+
version_it = it->find("nsolid");
194+
if (version_it != it->end() && version_it->is_string()) {
195+
std::string nsolid_version = version_it->get<std::string>();
196+
attrs.SetAttribute(kProcessRuntimeDescription,
197+
"N|Solid " + nsolid_version);
198+
}
199+
}
200+
201+
attrs.SetAttribute(kProcessRuntimeName, "nodejs");
202+
203+
it = info.find("cpuCores");
204+
if (it != info.end() && it->is_number_unsigned()) {
205+
attrs.SetAttribute("cpuCores", std::to_string(it->get<uint32_t>()));
206+
}
207+
208+
it = info.find("cpuModel");
209+
if (it != info.end() && it->is_string()) {
210+
attrs.SetAttribute(kHostCpuModelName, it->get<std::string>());
211+
}
212+
213+
it = info.find("processStart");
214+
if (it != info.end() && it->is_number_unsigned()) {
215+
std::string iso_time = ToIso8601(it->get<uint64_t>());
216+
if (!iso_time.empty()) {
217+
attrs.SetAttribute(kProcessCreationTime, std::move(iso_time));
218+
}
219+
}
220+
221+
it = info.find("tags");
222+
if (it != info.end() && it->is_array()) {
223+
std::string tags;
224+
for (const auto& tag : *it) {
225+
if (!tag.is_string()) continue;
226+
if (!tags.empty()) tags += ',';
227+
tags += tag.get<std::string>();
228+
}
229+
attrs.SetAttribute("tagsString", std::move(tags));
230+
}
231+
232+
return attrs;
233+
}
234+
235+
static std::shared_ptr<Resource> MergeResourceAttributes(
236+
const std::shared_ptr<Resource>& base,
237+
ResourceAttributes attrs) {
238+
auto resource_attributes = base->GetAttributes();
239+
if (resource_attributes.find(kServiceName) != resource_attributes.end() &&
240+
attrs.find(kServiceName) == attrs.end()) {
241+
attrs.SetAttribute(
242+
kServiceName,
243+
opentelemetry::nostd::get<std::string>(
244+
resource_attributes[kServiceName]));
245+
}
246+
auto overlay = std::make_shared<Resource>(Resource::Create(attrs));
247+
return std::make_shared<Resource>(base->Merge(*overlay));
248+
}
249+
250+
InstrumentationScope* GetScope() {
251+
static std::unique_ptr<InstrumentationScope> scope =
252+
InstrumentationScope::Create("nsolid", NODE_VERSION "+ns" NSOLID_VERSION);
253+
return scope.get();
254+
}
255+
256+
static void EnsureResourceInitializedLocked() {
257+
if (resource_g != nullptr) return;
258+
259+
json config = json::parse(nsolid::GetConfig(), nullptr, false);
260+
// assert because the runtime should never send me an invalid JSON config
261+
ASSERT(!config.is_discarded());
262+
auto it = config.find("app");
263+
ASSERT(it != config.end());
264+
ResourceAttributes attrs({
265+
{kServiceName, it->get<std::string>()},
266+
{kServiceInstanceId, nsolid::GetAgentId()}
267+
});
268+
269+
it = config.find("appVersion");
270+
if (it != config.end()) {
271+
attrs.SetAttribute(kServiceVersion, it->get<std::string>());
272+
}
273+
274+
resource_g = std::make_shared<Resource>(Resource::Create(attrs));
275+
}
276+
277+
std::shared_ptr<Resource> GetResource() {
278+
nsuv::ns_mutex::scoped_lock lock(ResourceMutex());
279+
EnsureResourceInitializedLocked();
280+
return resource_g;
281+
}
282+
283+
static void EnsureMetricsResourceInitializedLocked() {
284+
if (metrics_resource_g != nullptr) return;
285+
286+
EnsureResourceInitializedLocked();
287+
288+
json info = json::parse(nsolid::GetProcessInfo(), nullptr, false);
289+
ResourceAttributes attrs = GetMetadataResourceAttributes(info);
290+
metrics_resource_g = MergeResourceAttributes(resource_g, std::move(attrs));
291+
}
292+
293+
std::shared_ptr<Resource> GetMetricsResource() {
294+
nsuv::ns_mutex::scoped_lock lock(ResourceMutex());
295+
EnsureMetricsResourceInitializedLocked();
296+
return metrics_resource_g;
297+
}
298+
299+
std::shared_ptr<Resource> UpdateResource(ResourceAttributes&& attrs) {
300+
nsuv::ns_mutex::scoped_lock lock(ResourceMutex());
301+
EnsureResourceInitializedLocked();
302+
303+
ResourceAttributes metrics_attrs(attrs);
304+
resource_g = MergeResourceAttributes(resource_g, std::move(attrs));
305+
306+
if (metrics_resource_g != nullptr) {
307+
metrics_resource_g = MergeResourceAttributes(metrics_resource_g,
308+
std::move(metrics_attrs));
309+
}
310+
311+
return resource_g;
312+
}
73313

74314
// NOLINTNEXTLINE(runtime/references)
75315
static void add_counter(std::vector<MetricData>& metrics,
@@ -136,53 +376,6 @@ static void add_summary(std::vector<MetricData>& metrics,
136376
metrics.push_back(metric_data);
137377
}
138378

139-
InstrumentationScope* GetScope() {
140-
static std::unique_ptr<InstrumentationScope> scope =
141-
InstrumentationScope::Create("nsolid", NODE_VERSION "+ns" NSOLID_VERSION);
142-
return scope.get();
143-
}
144-
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;
165-
}
166-
167-
return resource_g.get();
168-
}
169-
170-
Resource* UpdateResource(ResourceAttributes&& attrs) {
171-
// First, get current kServiceName to avoid overwriting it with the default
172-
// value "unknown_service". (See Resource::Create() method in the SDK).
173-
auto resource = GetResource();
174-
auto attributes = resource->GetAttributes();
175-
if (attributes.find(kServiceName) != attributes.end() &&
176-
attrs.find(kServiceName) == attrs.end()) {
177-
attrs.SetAttribute(kServiceName,
178-
opentelemetry::nostd::get<std::string>(attributes[kServiceName]));
179-
}
180-
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();
184-
}
185-
186379
// NOLINTNEXTLINE(runtime/references)
187380
void fill_proc_metrics(std::vector<MetricData>& metrics,
188381
const ProcessMetrics::MetricsStor& stor,
@@ -250,7 +443,7 @@ NSOLID_PROCESS_METRICS_DOUBLE(V)
250443
if (prev_stor.user != stor.user || prev_stor.title != stor.title) {
251444
ResourceAttributes attrs = {
252445
{ kProcessOwner, stor.user },
253-
{ "process.title", stor.title },
446+
{ kProcessTitle, stor.title },
254447
};
255448

256449
USE(UpdateResource(std::move(attrs)));
@@ -370,7 +563,8 @@ void fill_log_recordable(LogsRecordable* recordable,
370563
nanoseconds(static_cast<uint64_t>(info.timestamp))));
371564
recordable->SetTimestamp(ts);
372565
recordable->SetObservedTimestamp(ts);
373-
recordable->SetResource(*GetResource());
566+
auto resource = GetResource();
567+
recordable->SetResource(*resource);
374568
recordable->SetInstrumentationScope(*GetScope());
375569
}
376570

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

511-
recordable->SetResource(*GetResource());
705+
auto resource = GetResource();
706+
recordable->SetResource(*resource);
512707
}
513708

514709
} // namespace otlp

0 commit comments

Comments
 (0)