Skip to content

Commit c1fee7f

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 resource attributes derived from process info to improve downstream metrics handling. Keep resource access thread-safe by returning shared resource snapshots guarded by nsuv::ns_mutex. Split metric resources from the common resource and invalidate the cached metrics resource whenever process info is updated so later exports rebuild it with fresh metadata. Route runtime info updates through EnvList::StoreInfo() and update tests to validate the expanded metric resource attributes, including dynamic updates through both the JS API and gRPC reconfigure paths.
1 parent ccdcd1f commit c1fee7f

11 files changed

Lines changed: 491 additions & 85 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.
@@ -941,7 +942,8 @@ void GrpcAgent::env_deletion_cb_(SharedEnvInst envinst,
941942
}
942943

943944
ResourceMetrics data;
944-
data.resource_ = otlp::GetResource();
945+
auto resource = otlp::GetMetricsResource();
946+
data.resource_ = resource.get();
945947
std::vector<MetricData> metrics;
946948

947949
ThreadMetricsStor stor;
@@ -1386,7 +1388,8 @@ void GrpcAgent::got_proc_metrics() {
13861388
std::vector<MetricData> metrics;
13871389
otlp::fill_proc_metrics(metrics, stor, proc_prev_stor_, false);
13881390
ResourceMetrics data;
1389-
data.resource_ = otlp::GetResource();
1391+
auto resource = otlp::GetMetricsResource();
1392+
data.resource_ = resource.get();
13901393
data.scope_metric_data_ =
13911394
std::vector<ScopeMetrics>{{otlp::GetScope(), metrics}};
13921395
auto result = metrics_exporter_->Export(data);

‎agents/otlp/src/otlp_common.cc‎

Lines changed: 253 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,235 @@ 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", 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+
void InvalidateMetricsResource() {
300+
nsuv::ns_mutex::scoped_lock lock(ResourceMutex());
301+
metrics_resource_g.reset();
302+
}
303+
304+
std::shared_ptr<Resource> UpdateResource(ResourceAttributes&& attrs) {
305+
nsuv::ns_mutex::scoped_lock lock(ResourceMutex());
306+
EnsureResourceInitializedLocked();
307+
308+
ResourceAttributes metrics_attrs(attrs);
309+
resource_g = MergeResourceAttributes(resource_g, std::move(attrs));
310+
311+
if (metrics_resource_g != nullptr) {
312+
metrics_resource_g = MergeResourceAttributes(metrics_resource_g,
313+
std::move(metrics_attrs));
314+
}
315+
316+
return resource_g;
317+
}
73318

74319
// NOLINTNEXTLINE(runtime/references)
75320
static void add_counter(std::vector<MetricData>& metrics,
@@ -136,53 +381,6 @@ static void add_summary(std::vector<MetricData>& metrics,
136381
metrics.push_back(metric_data);
137382
}
138383

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-
186384
// NOLINTNEXTLINE(runtime/references)
187385
void fill_proc_metrics(std::vector<MetricData>& metrics,
188386
const ProcessMetrics::MetricsStor& stor,
@@ -250,7 +448,7 @@ NSOLID_PROCESS_METRICS_DOUBLE(V)
250448
if (prev_stor.user != stor.user || prev_stor.title != stor.title) {
251449
ResourceAttributes attrs = {
252450
{ kProcessOwner, stor.user },
253-
{ "process.title", stor.title },
451+
{ kProcessTitle, stor.title },
254452
};
255453

256454
USE(UpdateResource(std::move(attrs)));
@@ -370,7 +568,8 @@ void fill_log_recordable(LogsRecordable* recordable,
370568
nanoseconds(static_cast<uint64_t>(info.timestamp))));
371569
recordable->SetTimestamp(ts);
372570
recordable->SetObservedTimestamp(ts);
373-
recordable->SetResource(*GetResource());
571+
auto resource = GetResource();
572+
recordable->SetResource(*resource);
374573
recordable->SetInstrumentationScope(*GetScope());
375574
}
376575

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

511-
recordable->SetResource(*GetResource());
710+
auto resource = GetResource();
711+
recordable->SetResource(*resource);
512712
}
513713

514714
} // namespace otlp

0 commit comments

Comments
 (0)