Update use of published_at_ns in process context C/C++ implementation + some small fixes (#66)

ivoanjo · web-flow · commit beb6a714e0e4 · 2026-03-10T14:59:54.000+01:00
**What does this PR do?**

This PR collects a number of fixes to the process context C/C++
implementation. I recommend reviewing it commit-by-commit (and let me
know if this is too annoying to review, I can break it down further).

The changes around "MFD_NOEXEC_SEAL on failure" and "Change publish
synchronization to use timestamp instead of signature" I haven't yet
pushed the change to the upstream spec, I'll do that next, but I wanted
to have a working version first :)

**Motivation:**

Improve implementation/spec :)

**Additional Notes:**

N/A

**How to test the change?**

The built-in decoder and otel_process_ctx_dump.sh have been updated to
match.
diff --git a/process-context/c-and-cpp/example_ctx.c b/process-context/c-and-cpp/example_ctx.c
@@ -39,6 +39,18 @@ void burn_cpu_for(int seconds) {
   getchar();
 }
 
+static void print_key_value_pairs(const char *label, const char **pairs) {
+  if (pairs && pairs[0] != NULL) {
+    printf(", %s=", label);
+    for (int i = 0; pairs[i] != NULL; i += 2) {
+      if (i > 0) printf(",");
+      printf("%s:%s", pairs[i], pairs[i + 1]);
+    }
+  } else {
+    printf(", %s=(none)", label);
+  }
+}
+
 bool read_and_print_ctx(const char* prefix) {
   otel_process_ctx_read_result result = otel_process_ctx_read();
 
@@ -60,27 +72,25 @@ bool read_and_print_ctx(const char* prefix) {
     result.data.telemetry_sdk_version
   );
 
-  if (result.data.resources && result.data.resources[0] != NULL) {
-    printf(", resources=");
-    for (int i = 0; result.data.resources[i] != NULL; i += 2) {
-      if (i > 0) printf(",");
-      printf("%s=%s", result.data.resources[i], result.data.resources[i + 1]);
-    }
-  } else {
-    printf(", resources=(none)");
-  }
+  print_key_value_pairs("resource_attributes", result.data.resource_attributes);
+  print_key_value_pairs("extra_attributes", result.data.extra_attributes);
   printf("\n");
 
   otel_process_ctx_read_drop(&result);
   return true;
 }
 
-const char *resources[] = {
+const char *resource_attributes[] = {
   "resource.key1", "resource.value1",
   "resource.key2", "resource.value2",
   NULL
 };
 
+const char *extra_attributes[] = {
+  "example_extra_attribute_foo", "example_extra_attribute_foo_value",
+  NULL
+};
+
 int update_and_fork(void) {
   printf("Burning CPU for 5 seconds...\n");
   burn_cpu_for(5);
@@ -94,7 +104,7 @@ int update_and_fork(void) {
     .telemetry_sdk_language = "c",
     .telemetry_sdk_version = "1.2.3",
     .telemetry_sdk_name = "example_ctx.c",
-    .resources = resources
+    .resource_attributes = resource_attributes
   };
 
   otel_process_ctx_result result = otel_process_ctx_publish(&update_data);
@@ -119,7 +129,7 @@ int update_and_fork(void) {
       .telemetry_sdk_language = "c",
       .telemetry_sdk_version = "1.2.3",
       .telemetry_sdk_name = "example_ctx.c",
-      .resources = NULL
+      .resource_attributes = NULL
     };
 
     result = otel_process_ctx_publish(&child_data);
@@ -172,7 +182,8 @@ int main(int argc, char* argv[]) {
     .telemetry_sdk_language = "c",
     .telemetry_sdk_version = "1.2.3",
     .telemetry_sdk_name = "example_ctx.c",
-    .resources = resources
+    .resource_attributes = resource_attributes,
+    .extra_attributes = extra_attributes
   };
 
   otel_process_ctx_result result = otel_process_ctx_publish(&data);
diff --git a/process-context/c-and-cpp/otel_process_ctx.c b/process-context/c-and-cpp/otel_process_ctx.c
@@ -42,7 +42,8 @@ static const otel_process_ctx_data empty_data = {
   .telemetry_sdk_language = NULL,
   .telemetry_sdk_version = NULL,
   .telemetry_sdk_name = NULL,
-  .resources = NULL
+  .resource_attributes = NULL,
+  .extra_attributes = NULL
 };
 
 #if (defined(OTEL_PROCESS_CTX_NOOP) && OTEL_PROCESS_CTX_NOOP) || !defined(__linux__)
@@ -75,10 +76,10 @@ static const otel_process_ctx_data empty_data = {
  * An outside-of-process reader will read this struct + otel_process_payload to get the data.
  */
 typedef struct __attribute__((packed, aligned(8))) {
-  char otel_process_ctx_signature[8];        // Always OTEL_CTX_SIGNATURE
+  char otel_process_ctx_signature[8];        // Always "OTEL_CTX"
   uint32_t otel_process_ctx_version;         // Always > 0, incremented when the data structure changes, currently v2
   uint32_t otel_process_payload_size;        // Always > 0, size of storage
-  uint64_t otel_process_ctx_published_at_ns; // Always > 0, timestamp from when the context was published in nanoseconds since epoch
+  uint64_t otel_process_ctx_published_at_ns; // Timestamp from when the context was published in nanoseconds since epoch. 0 during updates.
   char *otel_process_payload;                // Always non-null, points to the storage for the data; expected to be a protobuf map of string key/value pairs, null-terminated
 } otel_process_ctx_mapping;
 
@@ -144,6 +145,10 @@ otel_process_ctx_result otel_process_ctx_publish(const otel_process_ctx_data *da
   const ssize_t mapping_size = sizeof(otel_process_ctx_mapping);
   published_state.publisher_pid = getpid(); // This allows us to detect in forks that we shouldn't touch the mapping
   int fd = memfd_create("OTEL_CTX", MFD_CLOEXEC | MFD_ALLOW_SEALING | MFD_NOEXEC_SEAL);
+  if (fd < 0) {
+    // MFD_NOEXEC_SEAL is a newer flag; older kernels reject unknown flags, so let's retry without it
+    fd = memfd_create("OTEL_CTX", MFD_CLOEXEC | MFD_ALLOW_SEALING);
+  }
   bool failed_to_close_fd = false;
   if (fd >= 0) {
     // Try to create mapping from memfd
@@ -178,24 +183,24 @@ otel_process_ctx_result otel_process_ctx_publish(const otel_process_ctx_data *da
   }
 
   // Step: Populate the mapping
-  // The payload and any extra fields must come first and not be reordered with the signature by the compiler.
+  // The payload and any extra fields must come first and not be reordered with the published_at_ns by the compiler.
   *published_state.mapping = (otel_process_ctx_mapping) {
-    .otel_process_ctx_signature = {0}, // Set in "Step: Populate the signature into the mapping" below
+    .otel_process_ctx_signature = { 'O', 'T', 'E', 'L', '_', 'C', 'T', 'X' },
     .otel_process_ctx_version = 2,
     .otel_process_payload_size = payload_size,
-    .otel_process_ctx_published_at_ns = published_at_ns,
+    .otel_process_ctx_published_at_ns = 0, // Set in "Step: Populate the published_at_ns into the mapping" below
     .otel_process_payload = published_state.payload
   };
 
-  // Step: Synchronization - Mapping has been filled and is missing signature
-  // Make sure the initialization of the mapping + payload above does not get reordered with setting the signature below. Setting
-  // the signature is what tells an outside reader that the context is fully published.
+  // Step: Synchronization - Mapping has been filled and is missing published_at_ns
+  // Make sure the initialization of the mapping + payload above does not get reordered with setting the published_at_ns below. Setting
+  // the published_at_ns is what tells an outside reader that the context is fully published.
   atomic_thread_fence(memory_order_seq_cst);
 
-  // Step: Populate the signature into the mapping
-  // The signature must come last and not be reordered with the fields above by the compiler. After this step, external readers
-  // can read the signature and know that the payload is ready to be read.
-  memcpy(published_state.mapping->otel_process_ctx_signature, OTEL_CTX_SIGNATURE, sizeof(published_state.mapping->otel_process_ctx_signature));
+  // Step: Populate the published_at_ns into the mapping
+  // The published_at_ns must come last and not be reordered with the fields above by the compiler. After this step, external readers
+  // can read the published_at_ns and know that the payload is ready to be read.
+  published_state.mapping->otel_process_ctx_published_at_ns = published_at_ns;
 
   // Step: Attempt to name the mapping so outside readers can:
   // * Find it by name
@@ -242,6 +247,11 @@ static otel_process_ctx_result otel_process_ctx_update(uint64_t published_at_ns,
     return (otel_process_ctx_result) {.success = false, .error_message = "Unexpected: otel_process_ctx_data is NULL or context is not published (" __FILE__ ":" ADD_QUOTES(__LINE__) ")"};
   }
 
+  if (published_at_ns == published_state.mapping->otel_process_ctx_published_at_ns) {
+    // Advance published_at_ns to allow readers to detect the update
+    published_at_ns++;
+  }
+
   // Step: Prepare the new payload to be published
   // The payload SHOULD be ready and valid before trying to actually update the mapping.
   uint32_t payload_size = 0;
@@ -345,8 +355,8 @@ static void write_protobuf_tag(char **ptr, uint8_t field_number) {
   *(*ptr)++ = (char)((field_number << 3) | 2); // Field type is always 2 (LEN)
 }
 
-static void write_attribute(char **ptr, const char *key, const char *value) {
-  write_protobuf_tag(ptr, 1); // Resource.attributes (field 1)
+static void write_attribute(char **ptr, uint8_t field_number, const char *key, const char *value) {
+  write_protobuf_tag(ptr, field_number);
   write_protobuf_varint(ptr, protobuf_otel_keyvalue_string_size(key, value));
 
   // KeyValue
@@ -381,15 +391,20 @@ static otel_process_ctx_result otel_process_ctx_encode_protobuf_payload(char **o
   otel_process_ctx_result validation_result = validate_and_calculate_protobuf_payload_size(&pairs_size, (const char **) pairs);
   if (!validation_result.success) return validation_result;
 
-  size_t resources_pairs_size = 0;
-  if (data.resources != NULL) {
-    validation_result = validate_and_calculate_protobuf_payload_size(&resources_pairs_size, data.resources);
+  size_t resource_attributes_pairs_size = 0;
+  if (data.resource_attributes != NULL) {
+    validation_result = validate_and_calculate_protobuf_payload_size(&resource_attributes_pairs_size, data.resource_attributes);
     if (!validation_result.success) return validation_result;
   }
 
-  size_t resource_size = pairs_size + resources_pairs_size;
-  // ProcessContext wrapper: tag (1 byte) + resource length varint + resource content
-  size_t total_size = 1 + protobuf_varint_size(resource_size) + resource_size;
+  size_t extra_attributes_pairs_size = 0;
+  if (data.extra_attributes != NULL) {
+    validation_result = validate_and_calculate_protobuf_payload_size(&extra_attributes_pairs_size, data.extra_attributes);
+    if (!validation_result.success) return validation_result;
+  }
+
+  size_t resource_size = pairs_size + resource_attributes_pairs_size;
+  size_t total_size = protobuf_record_size(resource_size) + extra_attributes_pairs_size;
 
   char *encoded = (char *) calloc(total_size, 1);
   if (!encoded) {
@@ -402,11 +417,15 @@ static otel_process_ctx_result otel_process_ctx_encode_protobuf_payload(char **o
   write_protobuf_varint(&ptr, resource_size);
 
   for (size_t i = 0; pairs[i * 2] != NULL; i++) {
-    write_attribute(&ptr, pairs[i * 2], pairs[i * 2 + 1]);
+    write_attribute(&ptr, 1, pairs[i * 2], pairs[i * 2 + 1]);
   }
 
-  for (size_t i = 0; data.resources != NULL && data.resources[i * 2] != NULL; i++) {
-    write_attribute(&ptr, data.resources[i * 2], data.resources[i * 2 + 1]);
+  for (size_t i = 0; data.resource_attributes != NULL && data.resource_attributes[i * 2] != NULL; i++) {
+    write_attribute(&ptr, 1, data.resource_attributes[i * 2], data.resource_attributes[i * 2 + 1]);
+  }
+
+  for (size_t i = 0; data.extra_attributes != NULL && data.extra_attributes[i * 2] != NULL; i++) {
+    write_attribute(&ptr, 2, data.extra_attributes[i * 2], data.extra_attributes[i * 2 + 1]);
   }
 
   *out = encoded;
@@ -439,7 +458,7 @@ static otel_process_ctx_result otel_process_ctx_encode_protobuf_payload(char **o
     if (!fp) return result;
 
     while (fgets(line, sizeof(line), fp)) {
-      bool is_process_ctx = strstr(line, "[anon_shmem:OTEL_CTX]") != NULL || strstr(line, "/memfd:OTEL_CTX") != NULL;
+      bool is_process_ctx = strstr(line, "[anon_shmem:OTEL_CTX]") != NULL || strstr(line, "[anon:OTEL_CTX]") != NULL || strstr(line, "/memfd:OTEL_CTX") != NULL;
       if (is_process_ctx) {
         result = (otel_process_ctx_mapping *)parse_mapping_start(line);
         break;
@@ -495,6 +514,34 @@ static otel_process_ctx_result otel_process_ctx_encode_protobuf_payload(char **o
     return wire_type == 2; // We only need the LEN wire type for now
   }
 
+  // Reads an OTel KeyValue message (key string + AnyValue-wrapped string) into the provided buffers.
+  static bool read_protobuf_keyvalue(char **ptr, char *end_ptr, char *key_buffer, char *value_buffer) {
+    bool key_found = false;
+    bool value_found = false;
+
+    while (*ptr < end_ptr) {
+      uint8_t kv_field;
+      if (!read_protobuf_tag(ptr, end_ptr, &kv_field)) return false;
+
+      if (kv_field == 1) { // KeyValue.key
+        if (!read_protobuf_string(ptr, end_ptr, key_buffer)) return false;
+        key_found = true;
+      } else if (kv_field == 2) { // KeyValue.value (AnyValue)
+        uint16_t _any_len; // Unused, but we still need to consume + validate the varint
+        if (!read_protobuf_varint(ptr, end_ptr, &_any_len)) return false;
+        uint8_t any_field;
+        if (!read_protobuf_tag(ptr, end_ptr, &any_field)) return false;
+
+        if (any_field == 1) { // AnyValue.string_value
+          if (!read_protobuf_string(ptr, end_ptr, value_buffer)) return false;
+          value_found = true;
+        }
+      }
+    }
+
+    return key_found && value_found;
+  }
+
   // Simplified protobuf decoder to match the exact encoder above. If the protobuf data doesn't match the encoder, this will
   // return false.
   static bool otel_process_ctx_decode_payload(char *payload, uint32_t payload_size, otel_process_ctx_data *data_out, char *key_buffer, char *value_buffer) {
@@ -514,8 +561,13 @@ static otel_process_ctx_result otel_process_ctx_encode_protobuf_payload(char **o
 
     size_t resource_index = 0;
     size_t resource_capacity = 201; // Allocate space for 100 pairs + NULL terminator entry
-    data_out->resources = (const char **) calloc(resource_capacity, sizeof(char *));
-    if (data_out->resources == NULL) return false;
+    data_out->resource_attributes = (const char **) calloc(resource_capacity, sizeof(char *));
+    if (data_out->resource_attributes == NULL) return false;
+
+    size_t extra_attributes_index = 0;
+    size_t extra_attributes_capacity = 201; // Allocate space for 100 pairs + NULL terminator entry
+    data_out->extra_attributes = (const char **) calloc(extra_attributes_capacity, sizeof(char *));
+    if (data_out->extra_attributes == NULL) return false;
 
     while (ptr < resource_end) {
       uint8_t field_number;
@@ -526,31 +578,7 @@ static otel_process_ctx_result otel_process_ctx_encode_protobuf_payload(char **o
       char *kv_end = ptr + kv_len;
       if (kv_end > resource_end) return false;
 
-      bool key_found = false;
-      bool value_found = false;
-
-      // Parse KeyValue
-      while (ptr < kv_end) {
-        uint8_t kv_field;
-        if (!read_protobuf_tag(&ptr, kv_end, &kv_field)) return false;
-
-        if (kv_field == 1) { // KeyValue.key
-          if (!read_protobuf_string(&ptr, kv_end, key_buffer)) return false;
-          key_found = true;
-        } else if (kv_field == 2) { // KeyValue.value (AnyValue)
-          uint16_t _any_len; // Unused, but we still need to consume + validate the varint
-          if (!read_protobuf_varint(&ptr, kv_end, &_any_len)) return false;
-          uint8_t any_field;
-          if (!read_protobuf_tag(&ptr, kv_end, &any_field)) return false;
-
-          if (any_field == 1) { // AnyValue.string_value
-            if (!read_protobuf_string(&ptr, kv_end, value_buffer)) return false;
-            value_found = true;
-          }
-        }
-      }
-
-      if (!key_found || !value_found) return false;
+      if (!read_protobuf_keyvalue(&ptr, kv_end, key_buffer, value_buffer)) return false;
 
       char *value = strdup(value_buffer);
       if (!value) return false;
@@ -571,12 +599,35 @@ static otel_process_ctx_result otel_process_ctx_encode_protobuf_payload(char **o
           free(value);
           return false;
         }
-        data_out->resources[resource_index] = key;
-        data_out->resources[resource_index + 1] = value;
+        data_out->resource_attributes[resource_index] = key;
+        data_out->resource_attributes[resource_index + 1] = value;
         resource_index += 2;
       }
     }
 
+    while (ptr < end_ptr) {
+      uint8_t extra_ctx_field;
+      if (!read_protobuf_tag(&ptr, end_ptr, &extra_ctx_field) || extra_ctx_field != 2) return false;
+
+      uint16_t kv_len;
+      if (!read_protobuf_varint(&ptr, end_ptr, &kv_len)) return false;
+      char *kv_end = ptr + kv_len;
+      if (kv_end > end_ptr) return false;
+
+      if (!read_protobuf_keyvalue(&ptr, kv_end, key_buffer, value_buffer)) return false;
+
+      char *key = strdup(key_buffer);
+      char *value = strdup(value_buffer);
+      if (!key || !value || extra_attributes_index + 2 >= extra_attributes_capacity) {
+        free(key);
+        free(value);
+        return false;
+      }
+      data_out->extra_attributes[extra_attributes_index] = key;
+      data_out->extra_attributes[extra_attributes_index + 1] = value;
+      extra_attributes_index += 2;
+    }
+
     // Validate all required fields were found
     return data_out->deployment_environment_name != NULL &&
            data_out->service_instance_id != NULL &&
@@ -595,9 +646,13 @@ static otel_process_ctx_result otel_process_ctx_encode_protobuf_payload(char **o
     if (data.telemetry_sdk_language) free((void *)data.telemetry_sdk_language);
     if (data.telemetry_sdk_version) free((void *)data.telemetry_sdk_version);
     if (data.telemetry_sdk_name) free((void *)data.telemetry_sdk_name);
-    if (data.resources) {
-      for (int i = 0; data.resources[i] != NULL; i++) free((void *)data.resources[i]);
-      free((void *)data.resources);
+    if (data.resource_attributes) {
+      for (int i = 0; data.resource_attributes[i] != NULL; i++) free((void *)data.resource_attributes[i]);
+      free((void *)data.resource_attributes);
+    }
+    if (data.extra_attributes) {
+      for (int i = 0; data.extra_attributes[i] != NULL; i++) free((void *)data.extra_attributes[i]);
+      free((void *)data.extra_attributes);
     }
   }
 
diff --git a/process-context/c-and-cpp/otel_process_ctx.h b/process-context/c-and-cpp/otel_process_ctx.h
@@ -54,10 +54,14 @@ typedef struct {
   const char *telemetry_sdk_version;
   // https://opentelemetry.io/docs/specs/semconv/registry/attributes/telemetry/#telemetry-sdk-name
   const char *telemetry_sdk_name;
-  // Additional key/value pairs as resources https://opentelemetry.io/docs/specs/otel/resource/sdk/
-  // Can be NULL if no resources are needed; if non-NULL, this array MUST be terminated with a NULL entry.
+  // Additional key/value pairs as resource attributes https://opentelemetry.io/docs/specs/otel/resource/sdk/
+  // Can be NULL if no resource attributes are needed; if non-NULL, this array MUST be terminated with a NULL entry.
   // Every even entry is a key, every odd entry is a value (E.g. "key1", "value1", "key2", "value2", NULL).
-  const char **resources;
+  const char **resource_attributes;
+  // Additional key/value pairs as extra attributes (ProcessContext.extra_attributes in process_context.proto)
+  // Can be NULL if no extra attributes are needed; if non-NULL, this array MUST be terminated with a NULL entry.
+  // Every even entry is a key, every odd entry is a value (E.g. "key1", "value1", "key2", "value2", NULL).
+  const char **extra_attributes;
 } otel_process_ctx_data;
 
 /** Number of entries in the `otel_process_ctx_data` struct. Can be used to easily detect when the struct is updated. */
diff --git a/process-context/c-and-cpp/otel_process_ctx_dump.sh b/process-context/c-and-cpp/otel_process_ctx_dump.sh
@@ -27,7 +27,7 @@ if ! [[ "$pid" =~ ^[0-9]+$ ]]; then
 fi
 
 # Find the mapping by name
-if ! line="$(grep -F -m 1 -e '[anon_shmem:OTEL_CTX]' -e '/memfd:OTEL_CTX' "/proc/$pid/maps")"; then
+if ! line="$(grep -F -m 1 -e '[anon_shmem:OTEL_CTX]' -e '/memfd:OTEL_CTX' -e '[anon:OTEL_CTX]' "/proc/$pid/maps")"; then
   echo "No OTEL_CTX context found." >&2
   exit 1
 fi
