Skip to content

Submit worker count. Do not route metrics thru ext #3530

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cataphract wants to merge 5 commits into
base: master
Choose a base branch
from
Open

Submit worker count. Do not route metrics thru ext #3530

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
cc02664
Update libdatadog
cataphract Dec 12, 2025
1a96912
Submit worker count
cataphract Dec 12, 2025
c0cee56
appsec: avoid sending metrics to the extension and back
cataphract Dec 12, 2025
5a8f0d9
Remove extension submission of tel metrics gotten from helper
cataphract Dec 12, 2025
1d00e73
Fix tests
cataphract Dec 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 0 additions & 7 deletions appsec/src/extension/commands/client_init.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,13 +208,6 @@ static void _process_meta_and_metrics(

mpack_node_t metrics = mpack_node_array_at(root, 4);
dd_command_process_metrics(metrics, span);

// NOLINTNEXTLINE(cppcoreguidelines-avoid-magic-numbers,readability-magic-numbers)
if (mpack_node_array_length(root) >= 6) {
// NOLINTNEXTLINE(cppcoreguidelines-avoid-magic-numbers,readability-magic-numbers)
mpack_node_t tel_metrics = mpack_node_array_at(root, 5);
dd_command_process_telemetry_metrics(tel_metrics);
}
}

static dd_result _check_helper_version(mpack_node_t root)
Expand Down
6 changes: 5 additions & 1 deletion appsec/src/extension/commands/request_shutdown.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@ static const char *nullable _header_content_type_zend_array(
static const dd_command_spec _spec = {
.name = "request_shutdown",
.name_len = sizeof("request_shutdown") - 1,
.num_args = 3, // a map, api sec sampling key, and sidecar queue id
.num_args =
4, // a map, api sec sampling key, sidecar queue id, and input_truncated
.outgoing_cb = _request_pack,
.incoming_cb = dd_command_proc_resp_verd_span_data,
.config_features_cb = dd_command_process_config_features_unexpected,
Expand Down Expand Up @@ -101,6 +102,9 @@ static dd_result _request_pack(mpack_writer_t *nonnull w, void *nonnull ctx)
// 3.
mpack_write(w, dd_trace_get_sidecar_queue_id());

// 4.
mpack_write_bool(w, dd_msgpack_helpers_is_data_truncated());

return dd_success;
}

Expand Down
141 changes: 0 additions & 141 deletions appsec/src/extension/commands_helpers.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,6 @@
#include <mpack.h>
#include <stdatomic.h>

static const char WAF_REQUEST_METRIC[] = "waf.requests";
static const size_t WAF_REQUEST_METRIC_LEN = sizeof(WAF_REQUEST_METRIC) - 1;
static const char TRUNCATED_TAG[] = "input_truncated=true";
static const size_t TRUNCATED_TAG_LEN = sizeof(TRUNCATED_TAG);
static const char TAG_SEPARATOR = ',';
static const size_t TAG_SEPARATOR_LEN = sizeof(TAG_SEPARATOR);

typedef struct _dd_omsg {
zend_llist iovecs;
mpack_writer_t writer;
Expand Down Expand Up @@ -508,8 +501,6 @@ static void dd_command_process_settings(mpack_node_t root);
* 2: [force keep: bool]
* 3: [meta: map]
* 4: [metrics: map]
* 5: [telemetry metrics: map string ->
* array(array(value: double, tags: string)])
* )
*/
#define RESP_INDEX_ACTION_PARAMS 0
Expand All @@ -518,7 +509,6 @@ static void dd_command_process_settings(mpack_node_t root);
#define RESP_INDEX_SETTINGS 3
#define RESP_INDEX_SPAN_META 4
#define RESP_INDEX_SPAN_METRICS 5
#define RESP_INDEX_TELEMETRY_METRICS 6

dd_result dd_command_proc_resp_verd_span_data(
mpack_node_t root, void *unspecnull _ctx)
Expand Down Expand Up @@ -557,11 +547,6 @@ dd_result dd_command_proc_resp_verd_span_data(
dd_command_process_metrics(metrics, span);
}

if (mpack_node_array_length(root) >= RESP_INDEX_TELEMETRY_METRICS + 1) {
dd_command_process_telemetry_metrics(
mpack_node_array_at(root, RESP_INDEX_TELEMETRY_METRICS));
}

return res;
}

Expand Down Expand Up @@ -795,132 +780,6 @@ bool dd_command_process_metrics(mpack_node_t root, zend_object *nonnull span)
return true;
}

static void _handle_telemetry_metric(const char *nonnull key_str,
size_t key_len, double value, const char *nonnull tags_str,
size_t tags_len);

bool dd_command_process_telemetry_metrics(mpack_node_t metrics)
{
if (mpack_node_type(metrics) != mpack_type_map) {
return false;
}

if (!ddtrace_metric_register_buffer) {
mlog_g(dd_log_debug, "ddtrace_metric_register_buffer unavailable");
return true;
}

for (size_t i = 0; i < mpack_node_map_count(metrics); i++) {
mpack_node_t key = mpack_node_map_key_at(metrics, i);

const char *key_str = mpack_node_str(key);
if (!key_str) {
continue;
}

size_t key_len = mpack_node_strlen(key);
mpack_node_t arr_value = mpack_node_map_value_at(metrics, i);

for (size_t j = 0; j < mpack_node_array_length(arr_value); j++) {
mpack_node_t value = mpack_node_array_at(arr_value, j);
mpack_node_t dval_node = mpack_node_array_at(value, 0);
double dval = mpack_node_double(dval_node);

const char *tags_str = "";
char *modified_tags_str = NULL;
size_t tags_len = 0;
if (mpack_node_array_length(value) >= 2) {
mpack_node_t tags = mpack_node_array_at(value, 1);
tags_str = mpack_node_str(tags);
tags_len = mpack_node_strlen(tags);
}
if (mpack_node_error(metrics) != mpack_ok) {
break;
}
if (dd_msgpack_helpers_is_data_truncated() &&
WAF_REQUEST_METRIC_LEN == key_len &&
memcmp(WAF_REQUEST_METRIC, key_str, WAF_REQUEST_METRIC_LEN) ==
0) {
size_t separator = 0;
if (tags_len > 0) {
separator = TAG_SEPARATOR_LEN;
}
modified_tags_str =
emalloc(tags_len + TRUNCATED_TAG_LEN + 1 + separator);
if (modified_tags_str) {
memcpy(modified_tags_str, tags_str, tags_len);
if (separator > 0) {
modified_tags_str[tags_len] = TAG_SEPARATOR;
}
memcpy(modified_tags_str + tags_len + separator,
TRUNCATED_TAG, TRUNCATED_TAG_LEN);
tags_len += TRUNCATED_TAG_LEN + separator;
tags_str = modified_tags_str;
}
}

_handle_telemetry_metric(
key_str, key_len, dval, tags_str, tags_len);
if (modified_tags_str) {
efree(modified_tags_str);
}
}
}

return true;
}

static void _init_zstr(
zend_string *_Atomic *nonnull zstr, const char *nonnull str, size_t len)
{
zend_string *zstr_cur = atomic_load_explicit(zstr, memory_order_acquire);
if (zstr_cur != NULL) {
return;
}
zend_string *zstr_new = zend_string_init(str, len, 1);
if (atomic_compare_exchange_strong_explicit(zstr, &(zend_string *){NULL},
zstr_new, memory_order_release, memory_order_relaxed)) {
return;
}
zend_string_release(zstr_new);
}

// NOLINTNEXTLINE(bugprone-easily-swappable-parameters)
void _handle_telemetry_metric(const char *nonnull key_str, size_t key_len,
double value, const char *nonnull tags_str, size_t tags_len)
{
#define HANDLE_METRIC(name, type) \
do { \
if (key_len == LSTRLEN(name) && memcmp(key_str, name, key_len) == 0) { \
static zend_string *_Atomic key_zstr; \
_init_zstr(&key_zstr, name, LSTRLEN(name)); \
zend_string *tags_zstr = zend_string_init(tags_str, tags_len, 0); \
dd_telemetry_add_metric(key_zstr, value, tags_zstr, type); \
zend_string_release(tags_zstr); \
mlog_g(dd_log_debug, \
"Telemetry metric %.*s added with tags %.*s and value %f", \
(int)key_len, key_str, (int)tags_len, tags_str, value); \
return; \
} \
} while (0)

HANDLE_METRIC("waf.requests", DDTRACE_METRIC_TYPE_COUNT);
HANDLE_METRIC("waf.updates", DDTRACE_METRIC_TYPE_COUNT);
HANDLE_METRIC("waf.init", DDTRACE_METRIC_TYPE_COUNT);
HANDLE_METRIC("waf.config_errors", DDTRACE_METRIC_TYPE_COUNT);

HANDLE_METRIC("remote_config.first_pull", DDTRACE_METRIC_TYPE_GAUGE);
HANDLE_METRIC("remote_config.last_success", DDTRACE_METRIC_TYPE_GAUGE);

// Rasp
HANDLE_METRIC("rasp.timeout", DDTRACE_METRIC_TYPE_COUNT);
HANDLE_METRIC("rasp.rule.match", DDTRACE_METRIC_TYPE_COUNT);
HANDLE_METRIC("rasp.rule.eval", DDTRACE_METRIC_TYPE_COUNT);
HANDLE_METRIC("rasp.error", DDTRACE_METRIC_TYPE_COUNT);

mlog_g(dd_log_info, "Unknown telemetry metric %.*s", (int)key_len, key_str);
}

static void _dump_in_msg(
dd_log_level_t lvl, const char *nonnull data, size_t data_len)
{
Expand Down
1 change: 0 additions & 1 deletion appsec/src/extension/commands_helpers.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@ dd_result dd_command_proc_resp_verd_span_data(mpack_node_t root,
/* Common helpers */
void dd_command_process_meta(mpack_node_t root, zend_object *nonnull span);
bool dd_command_process_metrics(mpack_node_t root, zend_object *nonnull span);
bool dd_command_process_telemetry_metrics(mpack_node_t root);
dd_result dd_command_process_config_features(
mpack_node_t root, ATTR_UNUSED void *nullable ctx);
dd_result dd_command_process_config_features_unexpected(
Expand Down
73 changes: 41 additions & 32 deletions appsec/src/helper/client.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,12 @@
#include "action.hpp"
#include "client.hpp"
#include "exception.hpp"
#include "metrics.hpp"
#include "network/broker.hpp"
#include "network/proto.hpp"
#include "service.hpp"
#include "service_config.hpp"
#include "std_logging.hpp"
#include "telemetry.hpp"

using namespace std::chrono_literals;

Expand All @@ -27,9 +27,11 @@ namespace dds {
namespace {

void collect_metrics(network::request_shutdown::response &response,
service &service, std::optional<engine::context> &context);
service &service, std::optional<engine::context> &context,
const sidecar_settings &sc_settings);
void collect_metrics(network::client_init::response &response, service &service,
std::optional<engine::context> &context);
std::optional<engine::context> &context,
const sidecar_settings &sc_settings);

template <typename M, typename... Mrest>
// NOLINTNEXTLINE(google-runtime-references)
Expand Down Expand Up @@ -195,9 +197,9 @@ bool client::handle_command(const network::client_init::request &command)
response->status = has_errors ? "fail" : "ok";
response->errors = std::move(errors);

if (service_) {
if (!service_.is_empty()) {
// may be null in testing
collect_metrics(*response, *service_, context_);
collect_metrics(*response, *service_, context_, sc_settings_);
}

try {
Expand All @@ -218,7 +220,7 @@ bool client::handle_command(const network::client_init::request &command)

template <typename T> bool client::service_guard()
{
if (!service_) {
if (service_.is_empty()) {
// This implies a failed client_init, we can't continue.
SPDLOG_DEBUG("no service available on {}", T::name);
send_error_response(*broker_);
Expand Down Expand Up @@ -349,7 +351,7 @@ bool client::compute_client_status()
return request_enabled_;
}

if (service_ == nullptr) {
if (service_.is_empty()) {
request_enabled_ = false;
return request_enabled_;
}
Expand Down Expand Up @@ -470,7 +472,9 @@ bool client::handle_command(network::request_shutdown::request &command)
return false;
}

collect_metrics(*response, *service_, context_);
context_->set_input_truncated(command.input_truncated);

collect_metrics(*response, *service_, context_, sc_settings_);
service_->drain_logs(sc_settings_);

return send_message<network::request_shutdown>(response);
Expand Down Expand Up @@ -503,8 +507,6 @@ void client::update_settings(
rc_settings.shmem_path = std::string{rc_path};
}

sidecar_settings const current_sc_settings =
service_->get_sidecar_settings();
std::shared_ptr<service> new_service =
service_manager_->get_or_create_service(
*engine_settings_, rc_settings, telemetry_settings);
Expand Down Expand Up @@ -552,7 +554,9 @@ void client::run(worker::queue_consumer &q)
namespace {

struct request_metrics_submitter : public telemetry::telemetry_submitter {
request_metrics_submitter() = default;
request_metrics_submitter(service &svc, bool input_truncated)
: service_{svc}, input_truncated_{input_truncated}
{}
~request_metrics_submitter() override = default;
request_metrics_submitter(const request_metrics_submitter &) = delete;
request_metrics_submitter &operator=(
Expand All @@ -563,16 +567,18 @@ struct request_metrics_submitter : public telemetry::telemetry_submitter {
void submit_metric(std::string_view name, double value,
telemetry::telemetry_tags tags) override
{
std::string tags_s = tags.consume();
if (input_truncated_ && name == metrics::waf_requests) {
tags.add("input_truncated", "true");
}
SPDLOG_TRACE("submit_metric [req]: name={}, value={}, tags={}", name,
value, tags_s);
tel_metrics[name].emplace_back(value, std::move(tags_s));
value, tags);
service_.submit_request_metric(name, value, std::move(tags));
};
void submit_span_metric(std::string_view name, double value) override
{
SPDLOG_TRACE(
"submit_span_metric [req]: name={}, value={}", name, value);
metrics[name] = value;
span_metrics[name] = value;
};
void submit_span_meta(std::string_view name, std::string value) override
{
Expand All @@ -596,39 +602,42 @@ struct request_metrics_submitter : public telemetry::telemetry_submitter {
}

std::map<std::string, std::string> meta;
std::map<std::string_view, double> metrics;
std::unordered_map<std::string_view,
std::vector<std::pair<double, std::string>>>
tel_metrics;
std::map<std::string_view, double> span_metrics;
service &service_; // NOLINT
bool input_truncated_;
};

template <typename Response>
void collect_metrics_impl(Response &response, service &service,
std::optional<engine::context> &context)
std::optional<engine::context> &context,
const sidecar_settings &sc_settings)
{
request_metrics_submitter msubmitter{};
request_metrics_submitter msubmitter{
service, context ? context->get_input_truncated() : false};
if (context) {
// span metrics/meta go to msubmitter.span_metrics/meta and are sent to
// the extension; request telemetry metrics are routed to the service
// and sent to sidecar
context->get_metrics(msubmitter);
}
service.drain_metrics(
[&msubmitter](std::string_view name, double value, auto tags) {
msubmitter.submit_metric(name, value, std::move(tags));
});
msubmitter.metrics.merge(service.drain_legacy_metrics());

service.drain_metrics(sc_settings);
msubmitter.span_metrics.merge(service.drain_legacy_metrics());
msubmitter.meta.merge(service.drain_legacy_meta());
response.tel_metrics = std::move(msubmitter.tel_metrics);
response.meta = std::move(msubmitter.meta);
response.metrics = std::move(msubmitter.metrics);
response.metrics = std::move(msubmitter.span_metrics);
}
void collect_metrics(network::request_shutdown::response &response,
service &service, std::optional<engine::context> &context)
service &service, std::optional<engine::context> &context,
const sidecar_settings &sc_settings)
{
collect_metrics_impl(response, service, context);
collect_metrics_impl(response, service, context, sc_settings);
}
void collect_metrics(network::client_init::response &response, service &service,
std::optional<engine::context> &context)
std::optional<engine::context> &context,
const sidecar_settings &sc_settings)
{
collect_metrics_impl(response, service, context);
collect_metrics_impl(response, service, context, sc_settings);
}
} // namespace

Expand Down
Loading
Loading