diff --git a/lib/cfl/ARENA.md b/lib/cfl/ARENA.md index 6ee21566da1..3409f98900f 100644 --- a/lib/cfl/ARENA.md +++ b/lib/cfl/ARENA.md @@ -1,8 +1,9 @@ # CFL arena allocator `cfl_arena` is an optional allocator for CFL variants, arrays, key/value lists, -kvpairs, and owned SDS strings. It reduces allocator traffic when an application -constructs, mutates, and discards a complete object graph as one unit. +kvpairs, owned SDS strings, and arbitrary request-lifetime objects. It reduces +allocator traffic when an application constructs, mutates, and discards a +complete object graph as one unit. The normal CFL constructors remain heap-backed. Arena use is explicit and does not change existing callers. @@ -40,9 +41,20 @@ struct cfl_arena; struct cfl_arena *cfl_arena_create(size_t chunk_size); struct cfl_arena *cfl_arena_create_ex(size_t chunk_size, size_t large_object_threshold); +void cfl_arena_options_init(struct cfl_arena_options *options); +struct cfl_arena *cfl_arena_create_with_options( + const struct cfl_arena_options *options); void cfl_arena_destroy(struct cfl_arena *arena); void cfl_arena_reset(struct cfl_arena *arena); +void *cfl_arena_malloc(struct cfl_arena *arena, size_t size); +void *cfl_arena_calloc(struct cfl_arena *arena, + size_t count, size_t size); +void *cfl_arena_memdup(struct cfl_arena *arena, + const void *source, size_t size); +char *cfl_arena_strndup(struct cfl_arena *arena, + const char *source, size_t length); + size_t cfl_arena_bytes_reserved(struct cfl_arena *arena); size_t cfl_arena_bytes_used(struct cfl_arena *arena); size_t cfl_arena_large_object_threshold(struct cfl_arena *arena); @@ -57,6 +69,69 @@ Passing zero as `chunk_size` selects the default chunk size. With `cfl_arena_create_ex()`, a zero large-object threshold selects the default policy derived from the chunk size. +## Raw request-lifetime allocation + +Raw allocation supports objects that do not have CFL-specific constructors, +including temporary encoder trees: + +```c +struct request_state *state; +char *name; + +state = cfl_arena_calloc(arena, 1, sizeof(*state)); +name = cfl_arena_strndup(arena, input_name, input_name_length); +if (state == NULL || name == NULL) { + /* The arena remains valid and can still be reset or destroyed. */ +} +``` + +Raw pointers cannot be freed individually. They remain valid until the arena +is reset or destroyed. Returned pointers are aligned for CFL-supported +fundamental C types, including `long double`, pointers, and 64-bit integers. + +The raw allocation rules are: + +- `cfl_arena_malloc()` returns uninitialized storage. +- `cfl_arena_calloc()` checks multiplication overflow and zeroes the result. +- `cfl_arena_memdup()` copies an exact number of bytes. +- `cfl_arena_strndup()` appends a null terminator to the requested prefix. +- Zero-sized `malloc`, `calloc`, and `memdup` requests return `NULL`. +- `strndup` accepts a zero length and returns an allocated empty string. +- A null source, arithmetic overflow, invalid arena, or allocation failure + returns `NULL`. +- Failure leaves the arena usable and does not define `errno`. + +## Growth and allocator options + +Existing constructors retain fixed-size chunks. Optional geometric growth and +allocator callbacks are configured through an initialized options structure: + +```c +struct cfl_arena_options options; + +cfl_arena_options_init(&options); +options.chunk_size = 4096; +options.maximum_chunk_size = 65536; +options.malloc_fn = application_malloc; +options.free_fn = application_free; +options.allocator_context = application_context; + +arena = cfl_arena_create_with_options(&options); +``` + +The first normal chunk uses `chunk_size`. Later chunks double in size until +`maximum_chunk_size`; a request larger than the current chunk size receives a +dedicated chunk large enough for that request. A zero maximum selects fixed +growth, making the maximum equal to the initial chunk size. A maximum smaller +than the initial size is invalid. + +The callback pair is optional, but callers must provide both callbacks or +neither. Callbacks allocate and release the arena context, normal chunks, +external allocations, and cached external allocations. The allocation callback +must return storage with normal `malloc` alignment. CFL implements zeroing and +does not require `calloc` or `realloc` callbacks. `struct_size` must be set by +`cfl_arena_options_init()` so future CFL versions can extend the structure. + ## Arena-aware constructors The following constructors associate new values with an arena: @@ -168,9 +243,10 @@ allocation. ## Large values and external caching -Small and medium allocations come from arena chunks. Allocations at or above -the large-object threshold use separately tracked external storage so unusually -large values do not consume the remainder of a normal chunk. +Raw allocations and small CFL objects come from arena chunks. Raw requests +larger than the active chunk receive a dedicated chunk. The large-object +threshold applies to arena-aware owned SDS values, which use separately tracked +external storage so unusually large strings do not consume normal chunks. Reusable external buffers may remain cached after reset. Configure the maximum cached capacity with: diff --git a/lib/cfl/CHANGELOG.md b/lib/cfl/CHANGELOG.md index 8e72a65d0e0..37b1f89347f 100644 --- a/lib/cfl/CHANGELOG.md +++ b/lib/cfl/CHANGELOG.md @@ -2,6 +2,11 @@ This file records the notable changes in each CFL release. +## Unreleased + +- Added public request-lifetime arena allocation, duplication helpers, + allocator callbacks, and optional bounded geometric chunk growth. + ## 1.0.0 - 2026-07-11 The first stable CFL release establishes the variant, container, utility, and diff --git a/lib/cfl/CMakeLists.txt b/lib/cfl/CMakeLists.txt index 5484ff8f704..96188e4ea4f 100644 --- a/lib/cfl/CMakeLists.txt +++ b/lib/cfl/CMakeLists.txt @@ -158,7 +158,7 @@ endif() set(CPACK_PACKAGE_VERSION ${CFL_VERSION_STR}) set(CPACK_PACKAGE_NAME "cfl") set(CPACK_PACKAGE_RELEASE 1) -set(CPACK_PACKAGE_CONTACT "CFL Authors") +set(CPACK_PACKAGE_CONTACT "Eduardo Silva ") set(CPACK_PACKAGE_VENDOR "Fluent Project") set(CPACK_RESOURCE_FILE_LICENSE "${PROJECT_SOURCE_DIR}/LICENSE") set(CPACK_PACKAGING_INSTALL_PREFIX "/") diff --git a/lib/cfl/README.md b/lib/cfl/README.md index d7c452d4252..c73578dbdaf 100644 --- a/lib/cfl/README.md +++ b/lib/cfl/README.md @@ -25,7 +25,8 @@ smaller interface. - Intrusive lists and lightweight string key/value entries. - Portable 64-bit atomics and time helpers. - xxHash wrappers and CRC32C checksums. -- Optional arenas for allocation-heavy, bounded object graphs. +- Optional arenas for allocation-heavy, bounded object graphs and arbitrary + request-lifetime objects. - CMake support for embedding, installation, tests, and benchmarks. ## Core data structures diff --git a/lib/cfl/benchmarks/README.md b/lib/cfl/benchmarks/README.md index 021641c18c0..d4cb58f81fa 100644 --- a/lib/cfl/benchmarks/README.md +++ b/lib/cfl/benchmarks/README.md @@ -10,6 +10,18 @@ build-bench/benchmarks/cfl-benchmark-variant-arena heap 1000 1000 build-bench/benchmarks/cfl-benchmark-variant-arena arena 1000 1000 8192 ``` +Compare fixed 4 KiB chunks with optional 4-to-64 KiB geometric growth: + +```sh +build-bench/benchmarks/cfl-benchmark-variant-arena arena 1000 1000 4096 +build-bench/benchmarks/cfl-benchmark-variant-arena arena-grow 1000 1000 4096 65536 +``` + +The geometric mode exercises the public arena options used by request-lifetime +encoder workloads. Compare elapsed time, peak RSS, reserved bytes, used bytes, +and slack with a representative graph; fewer chunk allocations can trade CPU +time for retained capacity. + The tool reports elapsed time, peak RSS, glibc heap usage, and arena reserved/used bytes. Use `perf stat` for CPU and allocator-independent memory events: diff --git a/lib/cfl/benchmarks/arena.c b/lib/cfl/benchmarks/arena.c index 08140976ca1..f217378c96d 100644 --- a/lib/cfl/benchmarks/arena.c +++ b/lib/cfl/benchmarks/arena.c @@ -18,10 +18,7 @@ static uint64_t monotonic_nanoseconds(void) { - struct timespec now; - - timespec_get(&now, TIME_UTC); - return ((uint64_t) now.tv_sec * 1000000000ULL) + now.tv_nsec; + return cfl_time_now(); } static int build_heap(size_t entries) @@ -46,14 +43,24 @@ static int build_heap(size_t entries) } static int build_arena(size_t entries, size_t chunk_size, + size_t maximum_chunk_size, size_t *reserved, size_t *used) { struct cfl_arena *arena; + struct cfl_arena_options options; struct cfl_kvlist *list; size_t index; char key[32]; - arena = cfl_arena_create(chunk_size); + if (maximum_chunk_size == 0) { + arena = cfl_arena_create(chunk_size); + } + else { + cfl_arena_options_init(&options); + options.chunk_size = chunk_size; + options.maximum_chunk_size = maximum_chunk_size; + arena = cfl_arena_create_with_options(&options); + } if (arena == NULL) { return -1; } @@ -81,6 +88,7 @@ int main(int argc, char **argv) size_t iterations; size_t entries; size_t chunk_size; + size_t maximum_chunk_size; size_t iteration; size_t reserved; size_t used; @@ -97,13 +105,18 @@ int main(int argc, char **argv) iterations = argc > 2 ? strtoull(argv[2], NULL, 10) : 1000; entries = argc > 3 ? strtoull(argv[3], NULL, 10) : 1000; chunk_size = argc > 4 ? strtoull(argv[4], NULL, 10) : 8192; + maximum_chunk_size = argc > 5 ? strtoull(argv[5], NULL, 10) : 65536; reserved = 0; used = 0; start = monotonic_nanoseconds(); for (iteration = 0; iteration < iterations; iteration++) { - if (strcmp(mode, "arena") == 0) { - if (build_arena(entries, chunk_size, &reserved, &used) != 0) { + if (strcmp(mode, "arena") == 0 || + strcmp(mode, "arena-grow") == 0) { + if (build_arena(entries, chunk_size, + strcmp(mode, "arena-grow") == 0 ? + maximum_chunk_size : 0, + &reserved, &used) != 0) { return EXIT_FAILURE; } } @@ -113,7 +126,9 @@ int main(int argc, char **argv) } } else { - fprintf(stderr, "usage: %s heap|arena [iterations] [entries] [chunk-size]\n", + fprintf(stderr, + "usage: %s heap|arena|arena-grow [iterations] " + "[entries] [chunk-size] [maximum-chunk-size]\n", argv[0]); return EXIT_FAILURE; } @@ -132,7 +147,8 @@ int main(int argc, char **argv) printf(" heap_in_use=%zu heap_free=%zu", (size_t) memory.uordblks, (size_t) memory.fordblks); #endif - if (strcmp(mode, "arena") == 0) { + if (strcmp(mode, "arena") == 0 || + strcmp(mode, "arena-grow") == 0) { printf(" arena_reserved=%zu arena_used=%zu arena_slack=%zu", reserved, used, reserved - used); } diff --git a/lib/cfl/benchmarks/variant_mutable.c b/lib/cfl/benchmarks/variant_mutable.c index efe84a48809..62a9868b4ff 100644 --- a/lib/cfl/benchmarks/variant_mutable.c +++ b/lib/cfl/benchmarks/variant_mutable.c @@ -63,10 +63,7 @@ static size_t record_payload_size(const char *distribution, static uint64_t wall_nanoseconds(void) { - struct timespec now; - - timespec_get(&now, TIME_UTC); - return ((uint64_t) now.tv_sec * 1000000000ULL) + now.tv_nsec; + return cfl_time_now(); } static struct cfl_kvlist *create_record(struct cfl_arena *arena, diff --git a/lib/cfl/include/cfl/cfl_arena.h b/lib/cfl/include/cfl/cfl_arena.h index ba8eb21562b..2134df2a5b7 100644 --- a/lib/cfl/include/cfl/cfl_arena.h +++ b/lib/cfl/include/cfl/cfl_arena.h @@ -7,6 +7,19 @@ struct cfl_arena; +typedef void *(*cfl_arena_malloc_fn)(void *context, size_t size); +typedef void (*cfl_arena_free_fn)(void *context, void *pointer); + +struct cfl_arena_options { + size_t struct_size; + size_t chunk_size; + size_t maximum_chunk_size; + size_t large_object_threshold; + cfl_arena_malloc_fn malloc_fn; + cfl_arena_free_fn free_fn; + void *allocator_context; +}; + /* * Arena-created objects remain valid until the arena is reset or destroyed. * Reset and destroy invalidate every pointer allocated from the arena. @@ -20,8 +33,26 @@ struct cfl_arena; struct cfl_arena *cfl_arena_create(size_t chunk_size); struct cfl_arena *cfl_arena_create_ex(size_t chunk_size, size_t large_object_threshold); +void cfl_arena_options_init(struct cfl_arena_options *options); +struct cfl_arena *cfl_arena_create_with_options( + const struct cfl_arena_options *options); void cfl_arena_destroy(struct cfl_arena *arena); void cfl_arena_reset(struct cfl_arena *arena); + +/* + * Raw allocations are aligned for CFL-supported fundamental C types. They + * cannot be freed individually and remain valid until reset or destruction. + * A zero-sized or overflowing request returns NULL. Allocation failure leaves + * the arena valid and does not define errno. + */ +void *cfl_arena_malloc(struct cfl_arena *arena, size_t size); +void *cfl_arena_calloc(struct cfl_arena *arena, + size_t count, size_t size); +void *cfl_arena_memdup(struct cfl_arena *arena, + const void *source, size_t size); +char *cfl_arena_strndup(struct cfl_arena *arena, + const char *source, size_t length); + size_t cfl_arena_bytes_reserved(struct cfl_arena *arena); size_t cfl_arena_bytes_used(struct cfl_arena *arena); size_t cfl_arena_large_object_threshold(struct cfl_arena *arena); diff --git a/lib/cfl/src/cfl_arena.c b/lib/cfl/src/cfl_arena.c index bcdf0584545..62fa165fddd 100644 --- a/lib/cfl/src/cfl_arena.c +++ b/lib/cfl/src/cfl_arena.c @@ -50,14 +50,31 @@ struct cfl_arena { size_t external_cache_bytes; size_t external_cache_limit; size_t chunk_size; + size_t maximum_chunk_size; + size_t next_chunk_size; size_t bytes_reserved; size_t bytes_used; size_t large_object_threshold; void *free_variants; void *free_kvpairs; void *free_sds[CFL_ARENA_SDS_CLASS_COUNT]; + cfl_arena_malloc_fn malloc_fn; + cfl_arena_free_fn free_fn; + void *allocator_context; }; +static void *arena_default_malloc(void *context, size_t size) +{ + (void) context; + return malloc(size); +} + +static void arena_default_free(void *context, void *pointer) +{ + (void) context; + free(pointer); +} + static void arena_chunks_destroy(struct cfl_arena *arena) { struct cfl_arena_chunk *chunk; @@ -66,7 +83,7 @@ static void arena_chunks_destroy(struct cfl_arena *arena) chunk = arena->head; while (chunk != NULL) { next = chunk->next; - free(chunk); + arena->free_fn(arena->allocator_context, chunk); chunk = next; } @@ -88,7 +105,7 @@ static void arena_external_destroy(struct cfl_arena *arena) arena->bytes_reserved -= allocation->size + sizeof(struct cfl_arena_external); arena->bytes_used -= allocation->size; - free(allocation); + arena->free_fn(arena->allocator_context, allocation); allocation = next; } arena->external = NULL; @@ -99,7 +116,7 @@ static void arena_external_destroy(struct cfl_arena *arena) next = allocation->next; arena->bytes_reserved -= allocation->size + sizeof(struct cfl_arena_external); - free(allocation); + arena->free_fn(arena->allocator_context, allocation); allocation = next; } arena->external_cache[index] = NULL; @@ -112,7 +129,7 @@ static void arena_external_destroy(struct cfl_arena *arena) next = allocation->next; arena->bytes_reserved -= allocation->size + sizeof(struct cfl_arena_external); - free(allocation); + arena->free_fn(arena->allocator_context, allocation); allocation = next; } arena->external_exact_cache = NULL; @@ -125,19 +142,82 @@ struct cfl_arena *cfl_arena_create(size_t chunk_size) struct cfl_arena *cfl_arena_create_ex(size_t chunk_size, size_t large_object_threshold) +{ + struct cfl_arena_options options; + + cfl_arena_options_init(&options); + options.chunk_size = chunk_size; + options.large_object_threshold = large_object_threshold; + + return cfl_arena_create_with_options(&options); +} + +void cfl_arena_options_init(struct cfl_arena_options *options) +{ + if (options == NULL) { + return; + } + + memset(options, 0, sizeof(struct cfl_arena_options)); + options->struct_size = sizeof(struct cfl_arena_options); +} + +struct cfl_arena *cfl_arena_create_with_options( + const struct cfl_arena_options *options) { struct cfl_arena *arena; + size_t chunk_size; + size_t maximum_chunk_size; + size_t large_object_threshold; + cfl_arena_malloc_fn malloc_fn; + cfl_arena_free_fn free_fn; + void *allocator_context; + + if (options == NULL || + options->struct_size < + offsetof(struct cfl_arena_options, allocator_context) + + sizeof(options->allocator_context)) { + return NULL; + } + + if ((options->malloc_fn == NULL) != (options->free_fn == NULL)) { + return NULL; + } + + chunk_size = options->chunk_size; + maximum_chunk_size = options->maximum_chunk_size; + large_object_threshold = options->large_object_threshold; + malloc_fn = options->malloc_fn; + free_fn = options->free_fn; + allocator_context = options->allocator_context; if (chunk_size == 0) { chunk_size = CFL_ARENA_DEFAULT_CHUNK_SIZE; } + if (maximum_chunk_size == 0) { + maximum_chunk_size = chunk_size; + } + if (maximum_chunk_size < chunk_size) { + return NULL; + } + if (malloc_fn == NULL) { + malloc_fn = arena_default_malloc; + free_fn = arena_default_free; + allocator_context = NULL; + } - arena = calloc(1, sizeof(struct cfl_arena)); + arena = malloc_fn(allocator_context, sizeof(struct cfl_arena)); if (arena == NULL) { return NULL; } + memset(arena, 0, sizeof(struct cfl_arena)); arena->chunk_size = chunk_size; + arena->maximum_chunk_size = maximum_chunk_size; + arena->next_chunk_size = chunk_size; + arena->malloc_fn = malloc_fn; + arena->free_fn = free_fn; + arena->allocator_context = allocator_context; if (large_object_threshold == 0) { large_object_threshold = chunk_size / 2; if (large_object_threshold == 0) { @@ -162,7 +242,7 @@ void cfl_arena_destroy(struct cfl_arena *arena) arena_external_destroy(arena); arena_chunks_destroy(arena); - free(arena); + arena->free_fn(arena->allocator_context, arena); } void cfl_arena_reset(struct cfl_arena *arena) @@ -190,12 +270,13 @@ void cfl_arena_reset(struct cfl_arena *arena) arena->bytes_used = 0; arena->current = arena->head; + arena->next_chunk_size = arena->chunk_size; arena->free_variants = NULL; arena->free_kvpairs = NULL; memset(arena->free_sds, 0, sizeof(arena->free_sds)); } -void *cfl_arena_alloc(struct cfl_arena *arena, size_t size) +void *cfl_arena_malloc(struct cfl_arena *arena, size_t size) { struct cfl_arena_chunk *chunk; size_t alignment; @@ -229,7 +310,7 @@ void *cfl_arena_alloc(struct cfl_arena *arena, size_t size) chunk = chunk->next; } - capacity = arena->chunk_size; + capacity = arena->next_chunk_size; if (capacity < size) { capacity = size; } @@ -237,7 +318,8 @@ void *cfl_arena_alloc(struct cfl_arena *arena, size_t size) return NULL; } - chunk = malloc(sizeof(struct cfl_arena_chunk) + capacity); + chunk = arena->malloc_fn(arena->allocator_context, + sizeof(struct cfl_arena_chunk) + capacity); if (chunk == NULL) { return NULL; } @@ -251,9 +333,24 @@ void *cfl_arena_alloc(struct cfl_arena *arena, size_t size) sizeof(struct cfl_arena_chunk); arena->bytes_used += size; + if (capacity == arena->next_chunk_size && + arena->next_chunk_size < arena->maximum_chunk_size) { + if (arena->next_chunk_size > arena->maximum_chunk_size / 2) { + arena->next_chunk_size = arena->maximum_chunk_size; + } + else { + arena->next_chunk_size *= 2; + } + } + return chunk->data; } +void *cfl_arena_alloc(struct cfl_arena *arena, size_t size) +{ + return cfl_arena_malloc(arena, size); +} + void *cfl_arena_calloc(struct cfl_arena *arena, size_t count, size_t size) { @@ -265,7 +362,7 @@ void *cfl_arena_calloc(struct cfl_arena *arena, } total = count * size; - result = cfl_arena_alloc(arena, total); + result = cfl_arena_malloc(arena, total); if (result != NULL) { memset(result, 0, total); } @@ -273,6 +370,41 @@ void *cfl_arena_calloc(struct cfl_arena *arena, return result; } +void *cfl_arena_memdup(struct cfl_arena *arena, + const void *source, size_t size) +{ + void *result; + + if (source == NULL || size == 0) { + return NULL; + } + + result = cfl_arena_malloc(arena, size); + if (result != NULL) { + memcpy(result, source, size); + } + + return result; +} + +char *cfl_arena_strndup(struct cfl_arena *arena, + const char *source, size_t length) +{ + char *result; + + if (source == NULL || length == SIZE_MAX) { + return NULL; + } + + result = cfl_arena_malloc(arena, length + 1); + if (result != NULL) { + memcpy(result, source, length); + result[length] = '\0'; + } + + return result; +} + size_t cfl_arena_bytes_reserved(struct cfl_arena *arena) { return arena == NULL ? 0 : arena->bytes_reserved; @@ -310,7 +442,7 @@ void cfl_arena_external_cache_limit_set(struct cfl_arena *arena, arena->external_cache_bytes -= allocation->size; arena->bytes_reserved -= allocation->size + sizeof(struct cfl_arena_external); - free(allocation); + arena->free_fn(arena->allocator_context, allocation); } } @@ -321,7 +453,7 @@ void cfl_arena_external_cache_limit_set(struct cfl_arena *arena, arena->external_cache_bytes -= allocation->size; arena->bytes_reserved -= allocation->size + sizeof(struct cfl_arena_external); - free(allocation); + arena->free_fn(arena->allocator_context, allocation); } } @@ -401,8 +533,9 @@ void *cfl_arena_alloc_external(struct cfl_arena *arena, } if (allocation == NULL) { - allocation = malloc(sizeof(struct cfl_arena_external) + - allocation_size); + allocation = arena->malloc_fn( + arena->allocator_context, + sizeof(struct cfl_arena_external) + allocation_size); if (allocation == NULL) { return NULL; } @@ -466,7 +599,7 @@ void cfl_arena_free_external(struct cfl_arena *arena, else { arena->bytes_reserved -= allocation->size + sizeof(struct cfl_arena_external); - free(allocation); + arena->free_fn(arena->allocator_context, allocation); } } diff --git a/lib/cfl/src/cfl_arena_internal.h b/lib/cfl/src/cfl_arena_internal.h index df6e011ba43..e41fe3a943b 100644 --- a/lib/cfl/src/cfl_arena_internal.h +++ b/lib/cfl/src/cfl_arena_internal.h @@ -6,8 +6,6 @@ #include void *cfl_arena_alloc(struct cfl_arena *arena, size_t size); -void *cfl_arena_calloc(struct cfl_arena *arena, - size_t count, size_t size); void *cfl_arena_alloc_external(struct cfl_arena *arena, size_t size); void cfl_arena_free_external(struct cfl_arena *arena, diff --git a/lib/cfl/src/cfl_array.c b/lib/cfl/src/cfl_array.c index b327c9116d9..bac55ab67a9 100644 --- a/lib/cfl/src/cfl_array.c +++ b/lib/cfl/src/cfl_array.c @@ -49,7 +49,7 @@ struct cfl_array *cfl_array_create_in(struct cfl_arena *arena, array = malloc(sizeof(struct cfl_array)); } else { - array = cfl_arena_alloc(arena, sizeof(struct cfl_array)); + array = cfl_arena_malloc(arena, sizeof(struct cfl_array)); } if (array == NULL) { cfl_errno(); @@ -222,7 +222,7 @@ int cfl_array_append(struct cfl_array *array, tmp = realloc(array->entries, new_size); } else { - tmp = cfl_arena_alloc(array->arena, new_size); + tmp = cfl_arena_malloc(array->arena, new_size); if (tmp != NULL) { memcpy(tmp, array->entries, array->entry_count * sizeof(void *)); diff --git a/lib/cfl/src/cfl_kvlist.c b/lib/cfl/src/cfl_kvlist.c index d853eb7f3d0..204ab7744ea 100644 --- a/lib/cfl/src/cfl_kvlist.c +++ b/lib/cfl/src/cfl_kvlist.c @@ -99,7 +99,7 @@ struct cfl_kvlist *cfl_kvlist_create_in(struct cfl_arena *arena) list = malloc(sizeof(struct cfl_kvlist)); } else { - list = cfl_arena_alloc(arena, sizeof(struct cfl_kvlist)); + list = cfl_arena_malloc(arena, sizeof(struct cfl_kvlist)); } if (list == NULL) { cfl_report_runtime_error(); diff --git a/lib/cfl/src/cfl_sds.c b/lib/cfl/src/cfl_sds.c index 304e412882d..afe89c2a937 100644 --- a/lib/cfl/src/cfl_sds.c +++ b/lib/cfl/src/cfl_sds.c @@ -99,7 +99,7 @@ static cfl_sds_t sds_alloc(struct cfl_arena *arena, size_t size) buf = cfl_arena_alloc_external(arena, allocation_size); } else { - buf = cfl_arena_alloc(arena, allocation_size); + buf = cfl_arena_malloc(arena, allocation_size); } } if (!buf) { diff --git a/lib/cfl/tests/arena.c b/lib/cfl/tests/arena.c index 05cb7dee49b..494fa635088 100644 --- a/lib/cfl/tests/arena.c +++ b/lib/cfl/tests/arena.c @@ -17,6 +17,193 @@ struct test_alignment_probe { union test_max_align value; }; +struct test_allocator_context { + size_t allocation_count; + size_t free_count; + size_t fail_after; + size_t allocation_sizes[8]; +}; + +static void *test_allocator_malloc(void *data, size_t size) +{ + void *result; + struct test_allocator_context *context; + + context = data; + if (context->allocation_count >= context->fail_after) { + return NULL; + } + + result = malloc(size); + if (result != NULL) { + if (context->allocation_count < 8) { + context->allocation_sizes[context->allocation_count] = size; + } + context->allocation_count++; + } + + return result; +} + +static void test_allocator_free(void *data, void *pointer) +{ + struct test_allocator_context *context; + + context = data; + context->free_count++; + free(pointer); +} + +static void public_raw_allocations(void) +{ + unsigned char *zeroed; + unsigned char source[] = {0x01, 0x02, 0x03, 0x04}; + unsigned char *copy; + char *string; + void *pointer; + size_t alignment; + size_t index; + struct cfl_arena *arena; + + arena = cfl_arena_create(128); + TEST_CHECK(arena != NULL); + alignment = offsetof(struct test_alignment_probe, value); + + for (index = 1; index <= 64; index++) { + pointer = cfl_arena_malloc(arena, index); + TEST_CHECK(pointer != NULL); + TEST_CHECK((uintptr_t) pointer % alignment == 0); + } + + zeroed = cfl_arena_calloc(arena, 8, sizeof(unsigned char)); + TEST_CHECK(zeroed != NULL); + for (index = 0; index < 8; index++) { + TEST_CHECK(zeroed[index] == 0); + } + + copy = cfl_arena_memdup(arena, source, sizeof(source)); + TEST_CHECK(copy != NULL); + TEST_CHECK(memcmp(copy, source, sizeof(source)) == 0); + + string = cfl_arena_strndup(arena, "arena-data", 5); + TEST_CHECK(string != NULL); + TEST_CHECK(strcmp(string, "arena") == 0); + string = cfl_arena_strndup(arena, "", 0); + TEST_CHECK(string != NULL); + TEST_CHECK(string[0] == '\0'); + + cfl_arena_destroy(arena); +} + +static void public_raw_allocation_failures(void) +{ + size_t used; + struct cfl_arena *arena; + + arena = cfl_arena_create(128); + TEST_CHECK(arena != NULL); + used = cfl_arena_bytes_used(arena); + + TEST_CHECK(cfl_arena_malloc(NULL, 1) == NULL); + TEST_CHECK(cfl_arena_malloc(arena, 0) == NULL); + TEST_CHECK(cfl_arena_malloc(arena, SIZE_MAX) == NULL); + TEST_CHECK(cfl_arena_calloc(arena, SIZE_MAX, 2) == NULL); + TEST_CHECK(cfl_arena_calloc(arena, 0, 1) == NULL); + TEST_CHECK(cfl_arena_memdup(arena, NULL, 1) == NULL); + TEST_CHECK(cfl_arena_memdup(arena, "x", 0) == NULL); + TEST_CHECK(cfl_arena_strndup(arena, NULL, 0) == NULL); + TEST_CHECK(cfl_arena_strndup(arena, "x", SIZE_MAX) == NULL); + TEST_CHECK(cfl_arena_bytes_used(arena) == used); + + cfl_arena_destroy(arena); +} + +static void options_growth_and_callbacks(void) +{ + char payload[2048]; + cfl_sds_t value; + void *pointer; + struct cfl_arena *arena; + struct cfl_arena_options options; + struct test_allocator_context context; + + memset(&context, 0, sizeof(context)); + context.fail_after = SIZE_MAX; + cfl_arena_options_init(&options); + options.chunk_size = 64; + options.maximum_chunk_size = 256; + options.malloc_fn = test_allocator_malloc; + options.free_fn = test_allocator_free; + options.allocator_context = &context; + + arena = cfl_arena_create_with_options(&options); + TEST_CHECK(arena != NULL); + TEST_CHECK(context.allocation_count == 1); + + pointer = cfl_arena_malloc(arena, 48); + TEST_CHECK(pointer != NULL); + pointer = cfl_arena_malloc(arena, 48); + TEST_CHECK(pointer != NULL); + pointer = cfl_arena_malloc(arena, 96); + TEST_CHECK(pointer != NULL); + TEST_CHECK(context.allocation_count == 4); + TEST_CHECK(context.allocation_sizes[1] < context.allocation_sizes[2]); + TEST_CHECK(context.allocation_sizes[2] < context.allocation_sizes[3]); + + cfl_arena_reset(arena); + TEST_CHECK(cfl_arena_bytes_used(arena) == 0); + pointer = cfl_arena_malloc(arena, 48); + TEST_CHECK(pointer != NULL); + + memset(payload, 'x', sizeof(payload)); + value = cfl_sds_create_len_in(arena, payload, (int) sizeof(payload)); + TEST_CHECK(value != NULL); + cfl_sds_destroy(value); + + cfl_arena_destroy(arena); + TEST_CHECK(context.free_count == context.allocation_count); + + options.maximum_chunk_size = 32; + TEST_CHECK(cfl_arena_create_with_options(&options) == NULL); + options.maximum_chunk_size = 64; + options.free_fn = NULL; + TEST_CHECK(cfl_arena_create_with_options(&options) == NULL); + TEST_CHECK(cfl_arena_create_with_options(NULL) == NULL); + options.free_fn = test_allocator_free; + options.struct_size = offsetof(struct cfl_arena_options, + allocator_context); + TEST_CHECK(cfl_arena_create_with_options(&options) == NULL); +} + +static void callback_allocation_failure(void) +{ + void *pointer; + struct cfl_arena *arena; + struct cfl_arena_options options; + struct test_allocator_context context; + + memset(&context, 0, sizeof(context)); + context.fail_after = 0; + cfl_arena_options_init(&options); + options.chunk_size = 64; + options.malloc_fn = test_allocator_malloc; + options.free_fn = test_allocator_free; + options.allocator_context = &context; + TEST_CHECK(cfl_arena_create_with_options(&options) == NULL); + + context.fail_after = 1; + arena = cfl_arena_create_with_options(&options); + TEST_CHECK(arena != NULL); + TEST_CHECK(cfl_arena_malloc(arena, 1) == NULL); + TEST_CHECK(cfl_arena_bytes_used(arena) == 0); + + context.fail_after = SIZE_MAX; + pointer = cfl_arena_malloc(arena, 1); + TEST_CHECK(pointer != NULL); + cfl_arena_destroy(arena); + TEST_CHECK(context.free_count == context.allocation_count); +} + static void arena_build_and_destroy(void) { struct cfl_arena *arena; @@ -290,6 +477,9 @@ static void bound_external_rounding_and_cache(void) payload_size = 1024 * 1024; payload = malloc(payload_size); TEST_CHECK(payload != NULL); + if (payload == NULL) { + return; + } memset(payload, 'x', payload_size); arena = cfl_arena_create(8192); @@ -340,6 +530,10 @@ static void reclaim_failed_variant_construction(void) } TEST_LIST = { + {"public_raw_allocations", public_raw_allocations}, + {"public_raw_allocation_failures", public_raw_allocation_failures}, + {"options_growth_and_callbacks", options_growth_and_callbacks}, + {"callback_allocation_failure", callback_allocation_failure}, {"arena_build_and_destroy", arena_build_and_destroy}, {"arena_reset", arena_reset}, {"reject_cross_arena_values", reject_cross_arena_values}, diff --git a/lib/cfl/tests/installed_consumer/main.c b/lib/cfl/tests/installed_consumer/main.c index a65589a9e8d..1c806b66fc3 100644 --- a/lib/cfl/tests/installed_consumer/main.c +++ b/lib/cfl/tests/installed_consumer/main.c @@ -6,7 +6,9 @@ int main(void) { int result; + char *request_name; uint64_t hash; + struct cfl_arena *arena; struct cfl_array *array; result = cfl_init(); @@ -36,5 +38,18 @@ int main(void) return 1; } + arena = cfl_arena_create(0); + if (arena == NULL) { + return 1; + } + + request_name = cfl_arena_strndup(arena, "installed", 9); + if (request_name == NULL || strcmp(request_name, "installed") != 0) { + cfl_arena_destroy(arena); + return 1; + } + + cfl_arena_destroy(arena); + return 0; } diff --git a/src/opentelemetry/flb_opentelemetry_otlp_proto.c b/src/opentelemetry/flb_opentelemetry_otlp_proto.c index 93cdbd16960..cbdbca01194 100644 --- a/src/opentelemetry/flb_opentelemetry_otlp_proto.c +++ b/src/opentelemetry/flb_opentelemetry_otlp_proto.c @@ -30,6 +30,7 @@ #include #include +#include #include #include @@ -43,6 +44,17 @@ #define FLB_OTEL_LOGS_SCHEMA_KEY "schema" #define FLB_OTEL_LOGS_SCHEMA_OTLP "otlp" #define FLB_OTEL_LOGS_METADATA_KEY "otlp" +#define FLB_OTEL_PROTO_ARENA_INITIAL_CHUNK_SIZE 4096 +#define FLB_OTEL_PROTO_ARENA_MAX_CHUNK_SIZE 65536 + +struct otlp_proto_arena { + struct cfl_arena *backend; +}; + +/* + * Protobuf objects live until the request has been packed. Pointer arrays that + * grow with realloc remain heap-backed and are released separately. + */ struct otlp_proto_logs_scope_state { int64_t scope_id; @@ -61,6 +73,63 @@ struct otlp_proto_logs_resource_state { static msgpack_object *msgpack_map_get_object(msgpack_object_map *map, const char *key); +static void *otlp_proto_arena_malloc(void *context, size_t size) +{ + (void) context; + + return flb_malloc(size); +} + +static void otlp_proto_arena_free(void *context, void *pointer) +{ + (void) context; + + flb_free(pointer); +} + +static int otlp_proto_arena_init(struct otlp_proto_arena *arena) +{ + struct cfl_arena_options options; + + cfl_arena_options_init(&options); + options.chunk_size = FLB_OTEL_PROTO_ARENA_INITIAL_CHUNK_SIZE; + options.maximum_chunk_size = FLB_OTEL_PROTO_ARENA_MAX_CHUNK_SIZE; + options.malloc_fn = otlp_proto_arena_malloc; + options.free_fn = otlp_proto_arena_free; + + arena->backend = cfl_arena_create_with_options(&options); + if (arena->backend == NULL) { + return -1; + } + + return 0; +} + +static void otlp_proto_arena_destroy(struct otlp_proto_arena *arena) +{ + cfl_arena_destroy(arena->backend); + arena->backend = NULL; +} + +static void *otlp_proto_arena_alloc(struct otlp_proto_arena *arena, size_t size) +{ + return cfl_arena_malloc(arena->backend, size); +} + +static void *otlp_proto_arena_calloc(struct otlp_proto_arena *arena, + size_t count, + size_t size) +{ + return cfl_arena_calloc(arena->backend, count, size); +} + +static char *otlp_proto_arena_strndup(struct otlp_proto_arena *arena, + const char *input, + size_t length) +{ + return cfl_arena_strndup(arena->backend, input, length); +} + static void set_result(int *result, int value) { if (result != NULL) { @@ -249,15 +318,13 @@ static int msgpack_map_get_int64(msgpack_object_map *map, return -1; } -static void otlp_kvpair_destroy(Opentelemetry__Proto__Common__V1__KeyValue *kvpair); -static void otlp_any_value_destroy(Opentelemetry__Proto__Common__V1__AnyValue *value); -static void destroy_log_record(Opentelemetry__Proto__Logs__V1__LogRecord *record); - -static Opentelemetry__Proto__Common__V1__ArrayValue *otlp_array_value_initialize(size_t entry_count) +static Opentelemetry__Proto__Common__V1__ArrayValue *otlp_array_value_initialize( + struct otlp_proto_arena *arena, size_t entry_count) { Opentelemetry__Proto__Common__V1__ArrayValue *value; - value = flb_calloc(1, sizeof(Opentelemetry__Proto__Common__V1__ArrayValue)); + value = otlp_proto_arena_calloc(arena, 1, + sizeof(Opentelemetry__Proto__Common__V1__ArrayValue)); if (value == NULL) { return NULL; } @@ -265,10 +332,9 @@ static Opentelemetry__Proto__Common__V1__ArrayValue *otlp_array_value_initialize opentelemetry__proto__common__v1__array_value__init(value); if (entry_count > 0) { - value->values = flb_calloc(entry_count, - sizeof(Opentelemetry__Proto__Common__V1__AnyValue *)); + value->values = otlp_proto_arena_calloc(arena, entry_count, + sizeof(Opentelemetry__Proto__Common__V1__AnyValue *)); if (value->values == NULL) { - flb_free(value); return NULL; } @@ -278,11 +344,13 @@ static Opentelemetry__Proto__Common__V1__ArrayValue *otlp_array_value_initialize return value; } -static Opentelemetry__Proto__Common__V1__KeyValue *otlp_kvpair_value_initialize() +static Opentelemetry__Proto__Common__V1__KeyValue *otlp_kvpair_value_initialize( + struct otlp_proto_arena *arena) { Opentelemetry__Proto__Common__V1__KeyValue *value; - value = flb_calloc(1, sizeof(Opentelemetry__Proto__Common__V1__KeyValue)); + value = otlp_proto_arena_calloc(arena, 1, + sizeof(Opentelemetry__Proto__Common__V1__KeyValue)); if (value != NULL) { opentelemetry__proto__common__v1__key_value__init(value); } @@ -290,11 +358,13 @@ static Opentelemetry__Proto__Common__V1__KeyValue *otlp_kvpair_value_initialize( return value; } -static Opentelemetry__Proto__Common__V1__KeyValueList *otlp_kvlist_value_initialize(size_t entry_count) +static Opentelemetry__Proto__Common__V1__KeyValueList *otlp_kvlist_value_initialize( + struct otlp_proto_arena *arena, size_t entry_count) { Opentelemetry__Proto__Common__V1__KeyValueList *value; - value = flb_calloc(1, sizeof(Opentelemetry__Proto__Common__V1__KeyValueList)); + value = otlp_proto_arena_calloc(arena, 1, + sizeof(Opentelemetry__Proto__Common__V1__KeyValueList)); if (value == NULL) { return NULL; } @@ -302,10 +372,9 @@ static Opentelemetry__Proto__Common__V1__KeyValueList *otlp_kvlist_value_initial opentelemetry__proto__common__v1__key_value_list__init(value); if (entry_count > 0) { - value->values = flb_calloc(entry_count, - sizeof(Opentelemetry__Proto__Common__V1__KeyValue *)); + value->values = otlp_proto_arena_calloc(arena, entry_count, + sizeof(Opentelemetry__Proto__Common__V1__KeyValue *)); if (value->values == NULL) { - flb_free(value); return NULL; } @@ -315,12 +384,13 @@ static Opentelemetry__Proto__Common__V1__KeyValueList *otlp_kvlist_value_initial return value; } -static Opentelemetry__Proto__Common__V1__AnyValue *otlp_any_value_initialize(int data_type, - size_t entry_count) +static Opentelemetry__Proto__Common__V1__AnyValue *otlp_any_value_initialize( + struct otlp_proto_arena *arena, int data_type, size_t entry_count) { Opentelemetry__Proto__Common__V1__AnyValue *value; - value = flb_calloc(1, sizeof(Opentelemetry__Proto__Common__V1__AnyValue)); + value = otlp_proto_arena_calloc(arena, 1, + sizeof(Opentelemetry__Proto__Common__V1__AnyValue)); if (value == NULL) { return NULL; } @@ -352,20 +422,18 @@ static Opentelemetry__Proto__Common__V1__AnyValue *otlp_any_value_initialize(int else if (data_type == MSGPACK_OBJECT_ARRAY) { value->value_case = OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_ARRAY_VALUE; - value->array_value = otlp_array_value_initialize(entry_count); + value->array_value = otlp_array_value_initialize(arena, entry_count); if (value->array_value == NULL) { - flb_free(value); return NULL; } } else if (data_type == MSGPACK_OBJECT_MAP) { value->value_case = OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_KVLIST_VALUE; - value->kvlist_value = otlp_kvlist_value_initialize(entry_count); + value->kvlist_value = otlp_kvlist_value_initialize(arena, entry_count); if (value->kvlist_value == NULL) { - flb_free(value); return NULL; } } @@ -374,120 +442,32 @@ static Opentelemetry__Proto__Common__V1__AnyValue *otlp_any_value_initialize(int OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_BYTES_VALUE; } else { - flb_free(value); return NULL; } return value; } -static void otlp_kvarray_destroy(Opentelemetry__Proto__Common__V1__KeyValue **kvarray, - size_t entry_count) -{ - size_t index; - - if (kvarray == NULL) { - return; - } - - for (index = 0; index < entry_count; index++) { - if (kvarray[index] != NULL) { - otlp_kvpair_destroy(kvarray[index]); - } - } - - flb_free(kvarray); -} - -static void otlp_kvlist_destroy(Opentelemetry__Proto__Common__V1__KeyValueList *kvlist) -{ - size_t index; - - if (kvlist == NULL) { - return; - } - - for (index = 0; index < kvlist->n_values; index++) { - otlp_kvpair_destroy(kvlist->values[index]); - } - - flb_free(kvlist->values); - flb_free(kvlist); -} - -static void otlp_array_destroy(Opentelemetry__Proto__Common__V1__ArrayValue *array) -{ - size_t index; - - if (array == NULL) { - return; - } - - for (index = 0; index < array->n_values; index++) { - otlp_any_value_destroy(array->values[index]); - } - - flb_free(array->values); - flb_free(array); -} - -static void otlp_any_value_destroy(Opentelemetry__Proto__Common__V1__AnyValue *value) -{ - if (value == NULL) { - return; - } - - if (value->value_case == - OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_STRING_VALUE) { - flb_free(value->string_value); - } - else if (value->value_case == - OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_ARRAY_VALUE) { - otlp_array_destroy(value->array_value); - } - else if (value->value_case == - OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_KVLIST_VALUE) { - otlp_kvlist_destroy(value->kvlist_value); - } - else if (value->value_case == - OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_BYTES_VALUE) { - flb_free(value->bytes_value.data); - } - - flb_free(value); -} - -static void otlp_kvpair_destroy(Opentelemetry__Proto__Common__V1__KeyValue *kvpair) -{ - if (kvpair == NULL) { - return; - } - - flb_free(kvpair->key); - otlp_any_value_destroy(kvpair->value); - flb_free(kvpair); -} - static Opentelemetry__Proto__Common__V1__AnyValue *msgpack_object_to_otlp_any_value( - msgpack_object *object); + struct otlp_proto_arena *arena, msgpack_object *object); static Opentelemetry__Proto__Common__V1__AnyValue *msgpack_array_to_otlp_any_value( - msgpack_object *object) + struct otlp_proto_arena *arena, msgpack_object *object) { size_t index; Opentelemetry__Proto__Common__V1__AnyValue *entry; Opentelemetry__Proto__Common__V1__AnyValue *value; - value = otlp_any_value_initialize(MSGPACK_OBJECT_ARRAY, + value = otlp_any_value_initialize(arena, MSGPACK_OBJECT_ARRAY, object->via.array.size); if (value == NULL) { return NULL; } for (index = 0; index < object->via.array.size; index++) { - entry = msgpack_object_to_otlp_any_value(&object->via.array.ptr[index]); + entry = msgpack_object_to_otlp_any_value(arena, + &object->via.array.ptr[index]); if (entry == NULL) { - otlp_any_value_destroy(value); return NULL; } @@ -498,30 +478,28 @@ static Opentelemetry__Proto__Common__V1__AnyValue *msgpack_array_to_otlp_any_val } static Opentelemetry__Proto__Common__V1__KeyValue *msgpack_kv_to_otlp_any_value( - struct msgpack_object_kv *input_pair) + struct otlp_proto_arena *arena, struct msgpack_object_kv *input_pair) { Opentelemetry__Proto__Common__V1__KeyValue *kv; - kv = otlp_kvpair_value_initialize(); + kv = otlp_kvpair_value_initialize(arena); if (kv == NULL) { return NULL; } if (input_pair->key.type != MSGPACK_OBJECT_STR) { - flb_free(kv); return NULL; } - kv->key = flb_strndup(input_pair->key.via.str.ptr, input_pair->key.via.str.size); + kv->key = otlp_proto_arena_strndup(arena, + input_pair->key.via.str.ptr, + input_pair->key.via.str.size); if (kv->key == NULL) { - flb_free(kv); return NULL; } - kv->value = msgpack_object_to_otlp_any_value(&input_pair->val); + kv->value = msgpack_object_to_otlp_any_value(arena, &input_pair->val); if (kv->value == NULL) { - flb_free(kv->key); - flb_free(kv); return NULL; } @@ -529,7 +507,7 @@ static Opentelemetry__Proto__Common__V1__KeyValue *msgpack_kv_to_otlp_any_value( } static Opentelemetry__Proto__Common__V1__KeyValue **msgpack_map_to_otlp_kvarray( - msgpack_object *object, size_t *entry_count) + struct otlp_proto_arena *arena, msgpack_object *object, size_t *entry_count) { size_t index; Opentelemetry__Proto__Common__V1__KeyValue **result; @@ -540,17 +518,17 @@ static Opentelemetry__Proto__Common__V1__KeyValue **msgpack_map_to_otlp_kvarray( return NULL; } - result = flb_calloc(*entry_count, - sizeof(Opentelemetry__Proto__Common__V1__KeyValue *)); + result = otlp_proto_arena_calloc(arena, *entry_count, + sizeof(Opentelemetry__Proto__Common__V1__KeyValue *)); if (result == NULL) { *entry_count = 0; return NULL; } for (index = 0; index < *entry_count; index++) { - result[index] = msgpack_kv_to_otlp_any_value(&object->via.map.ptr[index]); + result[index] = msgpack_kv_to_otlp_any_value(arena, + &object->via.map.ptr[index]); if (result[index] == NULL) { - otlp_kvarray_destroy(result, index); *entry_count = 0; return NULL; } @@ -560,21 +538,21 @@ static Opentelemetry__Proto__Common__V1__KeyValue **msgpack_map_to_otlp_kvarray( } static Opentelemetry__Proto__Common__V1__AnyValue *msgpack_map_to_otlp_any_value( - msgpack_object *object) + struct otlp_proto_arena *arena, msgpack_object *object) { size_t index; Opentelemetry__Proto__Common__V1__KeyValue *entry; Opentelemetry__Proto__Common__V1__AnyValue *value; - value = otlp_any_value_initialize(MSGPACK_OBJECT_MAP, object->via.map.size); + value = otlp_any_value_initialize(arena, MSGPACK_OBJECT_MAP, + object->via.map.size); if (value == NULL) { return NULL; } for (index = 0; index < object->via.map.size; index++) { - entry = msgpack_kv_to_otlp_any_value(&object->via.map.ptr[index]); + entry = msgpack_kv_to_otlp_any_value(arena, &object->via.map.ptr[index]); if (entry == NULL) { - otlp_any_value_destroy(value); return NULL; } @@ -585,7 +563,7 @@ static Opentelemetry__Proto__Common__V1__AnyValue *msgpack_map_to_otlp_any_value } static Opentelemetry__Proto__Common__V1__AnyValue *msgpack_object_to_otlp_any_value( - msgpack_object *object) + struct otlp_proto_arena *arena, msgpack_object *object) { Opentelemetry__Proto__Common__V1__AnyValue *value; @@ -597,21 +575,20 @@ static Opentelemetry__Proto__Common__V1__AnyValue *msgpack_object_to_otlp_any_va switch (object->type) { case MSGPACK_OBJECT_NIL: - value = otlp_any_value_initialize(MSGPACK_OBJECT_NIL, 0); + value = otlp_any_value_initialize(arena, MSGPACK_OBJECT_NIL, 0); break; case MSGPACK_OBJECT_BOOLEAN: - value = otlp_any_value_initialize(MSGPACK_OBJECT_BOOLEAN, 0); + value = otlp_any_value_initialize(arena, MSGPACK_OBJECT_BOOLEAN, 0); if (value != NULL) { value->bool_value = object->via.boolean; } break; case MSGPACK_OBJECT_POSITIVE_INTEGER: case MSGPACK_OBJECT_NEGATIVE_INTEGER: - value = otlp_any_value_initialize(object->type, 0); + value = otlp_any_value_initialize(arena, object->type, 0); if (value != NULL) { if (object->type == MSGPACK_OBJECT_POSITIVE_INTEGER) { if (object->via.u64 > INT64_MAX) { - otlp_any_value_destroy(value); value = NULL; break; } @@ -625,43 +602,44 @@ static Opentelemetry__Proto__Common__V1__AnyValue *msgpack_object_to_otlp_any_va break; case MSGPACK_OBJECT_FLOAT32: case MSGPACK_OBJECT_FLOAT64: - value = otlp_any_value_initialize(object->type, 0); + value = otlp_any_value_initialize(arena, object->type, 0); if (value != NULL) { value->double_value = object->via.f64; } break; case MSGPACK_OBJECT_STR: - value = otlp_any_value_initialize(MSGPACK_OBJECT_STR, 0); + value = otlp_any_value_initialize(arena, MSGPACK_OBJECT_STR, 0); if (value != NULL) { - value->string_value = flb_strndup(object->via.str.ptr, - object->via.str.size); + value->string_value = otlp_proto_arena_strndup(arena, + object->via.str.ptr, + object->via.str.size); if (value->string_value == NULL) { - otlp_any_value_destroy(value); value = NULL; } } break; case MSGPACK_OBJECT_BIN: - value = otlp_any_value_initialize(MSGPACK_OBJECT_BIN, 0); + value = otlp_any_value_initialize(arena, MSGPACK_OBJECT_BIN, 0); if (value != NULL) { value->bytes_value.len = object->via.bin.size; - value->bytes_value.data = flb_malloc(object->via.bin.size); - if (value->bytes_value.data == NULL) { - otlp_any_value_destroy(value); - value = NULL; + if (object->via.bin.size == 0) { + value->bytes_value.data = cfl_arena_malloc(arena->backend, 1); } else { - memcpy(value->bytes_value.data, - object->via.bin.ptr, - object->via.bin.size); + value->bytes_value.data = cfl_arena_memdup(arena->backend, + object->via.bin.ptr, + object->via.bin.size); + } + if (value->bytes_value.data == NULL) { + value = NULL; } } break; case MSGPACK_OBJECT_ARRAY: - value = msgpack_array_to_otlp_any_value(object); + value = msgpack_array_to_otlp_any_value(arena, object); break; case MSGPACK_OBJECT_MAP: - value = msgpack_map_to_otlp_any_value(object); + value = msgpack_map_to_otlp_any_value(arena, object); break; default: break; @@ -759,11 +737,13 @@ static msgpack_object *find_log_body_candidate(msgpack_object *body, return body; } -static int append_kvarrays(Opentelemetry__Proto__Common__V1__KeyValue ***base, +static int append_kvarrays(struct otlp_proto_arena *arena, + Opentelemetry__Proto__Common__V1__KeyValue ***base, size_t *base_count, Opentelemetry__Proto__Common__V1__KeyValue **extra, size_t extra_count) { + size_t total_count; Opentelemetry__Proto__Common__V1__KeyValue **tmp; if (extra == NULL || extra_count == 0) { @@ -776,23 +756,34 @@ static int append_kvarrays(Opentelemetry__Proto__Common__V1__KeyValue ***base, return 0; } - tmp = flb_realloc(*base, - sizeof(Opentelemetry__Proto__Common__V1__KeyValue *) * - (*base_count + extra_count)); + if (*base_count > SIZE_MAX - extra_count) { + return -1; + } + + total_count = *base_count + extra_count; + if (total_count > + SIZE_MAX / sizeof(Opentelemetry__Proto__Common__V1__KeyValue *)) { + return -1; + } + + tmp = otlp_proto_arena_alloc(arena, + sizeof(Opentelemetry__Proto__Common__V1__KeyValue *) * total_count); if (tmp == NULL) { return -1; } - *base = tmp; - memcpy(*base + *base_count, extra, + memcpy(tmp, *base, + sizeof(Opentelemetry__Proto__Common__V1__KeyValue *) * *base_count); + memcpy(tmp + *base_count, extra, sizeof(Opentelemetry__Proto__Common__V1__KeyValue *) * extra_count); + *base = tmp; *base_count += extra_count; - flb_free(extra); return 0; } static int msgpack_map_to_otlp_kvarray_filtered( + struct otlp_proto_arena *arena, msgpack_object_map *map, const char *ignored_key, size_t ignored_key_length, @@ -801,10 +792,8 @@ static int msgpack_map_to_otlp_kvarray_filtered( { size_t index; size_t count; - Opentelemetry__Proto__Common__V1__KeyValue **tmp; Opentelemetry__Proto__Common__V1__KeyValue **values; - values = NULL; count = 0; for (index = 0; index < map->size; index++) { @@ -817,20 +806,36 @@ static int msgpack_map_to_otlp_kvarray_filtered( continue; } - tmp = flb_realloc(values, - sizeof(Opentelemetry__Proto__Common__V1__KeyValue *) * - (count + 1)); - if (tmp == NULL) { - otlp_kvarray_destroy(values, count); - *out_values = NULL; - *out_count = 0; - return -1; + count++; + } + + if (count == 0) { + *out_values = NULL; + *out_count = 0; + return 0; + } + + values = otlp_proto_arena_calloc(arena, count, + sizeof(Opentelemetry__Proto__Common__V1__KeyValue *)); + if (values == NULL) { + *out_values = NULL; + *out_count = 0; + return -1; + } + + count = 0; + for (index = 0; index < map->size; index++) { + if (ignored_key != NULL && + map->ptr[index].key.type == MSGPACK_OBJECT_STR && + map->ptr[index].key.via.str.size == ignored_key_length && + strncmp(map->ptr[index].key.via.str.ptr, + ignored_key, + ignored_key_length) == 0) { + continue; } - values = tmp; - values[count] = msgpack_kv_to_otlp_any_value(&map->ptr[index]); + values[count] = msgpack_kv_to_otlp_any_value(arena, &map->ptr[index]); if (values[count] == NULL) { - otlp_kvarray_destroy(values, count); *out_values = NULL; *out_count = 0; return -1; @@ -846,6 +851,7 @@ static int msgpack_map_to_otlp_kvarray_filtered( } static int log_record_set_body_and_attributes( + struct otlp_proto_arena *arena, Opentelemetry__Proto__Logs__V1__LogRecord *record, struct flb_log_event *event, const char **logs_body_keys, @@ -866,7 +872,7 @@ static int log_record_set_body_and_attributes( &matched_key, &matched_key_length); - record->body = msgpack_object_to_otlp_any_value(candidate); + record->body = msgpack_object_to_otlp_any_value(arena, candidate); if (candidate != NULL && record->body == NULL) { return -1; } @@ -875,7 +881,8 @@ static int log_record_set_body_and_attributes( matched_key != NULL && event->body != NULL && event->body->type == MSGPACK_OBJECT_MAP) { - if (msgpack_map_to_otlp_kvarray_filtered(&event->body->via.map, + if (msgpack_map_to_otlp_kvarray_filtered(arena, + &event->body->via.map, matched_key, matched_key_length, &attributes, @@ -883,11 +890,11 @@ static int log_record_set_body_and_attributes( return -1; } - if (append_kvarrays(&record->attributes, + if (append_kvarrays(arena, + &record->attributes, &record->n_attributes, attributes, attribute_count) != 0) { - otlp_kvarray_destroy(attributes, attribute_count); return -1; } } @@ -896,6 +903,7 @@ static int log_record_set_body_and_attributes( } static int add_msgpack_attributes_to_resource( + struct otlp_proto_arena *arena, Opentelemetry__Proto__Resource__V1__Resource *resource, msgpack_object *resource_object) { @@ -907,7 +915,7 @@ static int add_msgpack_attributes_to_resource( field = msgpack_map_get_object(&resource_object->via.map, "attributes"); if (field != NULL && field->type == MSGPACK_OBJECT_MAP) { - resource->attributes = msgpack_map_to_otlp_kvarray(field, + resource->attributes = msgpack_map_to_otlp_kvarray(arena, field, &resource->n_attributes); if (field->via.map.size > 0 && resource->attributes == NULL) { return -1; @@ -930,6 +938,7 @@ static int add_msgpack_attributes_to_resource( } static int add_msgpack_scope_fields( + struct otlp_proto_arena *arena, Opentelemetry__Proto__Common__V1__InstrumentationScope *scope, msgpack_object *scope_object) { @@ -941,7 +950,9 @@ static int add_msgpack_scope_fields( field = msgpack_map_get_object(&scope_object->via.map, "name"); if (field != NULL && field->type == MSGPACK_OBJECT_STR) { - scope->name = flb_strndup(field->via.str.ptr, field->via.str.size); + scope->name = otlp_proto_arena_strndup(arena, + field->via.str.ptr, + field->via.str.size); if (scope->name == NULL) { return -1; } @@ -949,7 +960,9 @@ static int add_msgpack_scope_fields( field = msgpack_map_get_object(&scope_object->via.map, "version"); if (field != NULL && field->type == MSGPACK_OBJECT_STR) { - scope->version = flb_strndup(field->via.str.ptr, field->via.str.size); + scope->version = otlp_proto_arena_strndup(arena, + field->via.str.ptr, + field->via.str.size); if (scope->version == NULL) { return -1; } @@ -957,7 +970,7 @@ static int add_msgpack_scope_fields( field = msgpack_map_get_object(&scope_object->via.map, "attributes"); if (field != NULL && field->type == MSGPACK_OBJECT_MAP) { - scope->attributes = msgpack_map_to_otlp_kvarray(field, + scope->attributes = msgpack_map_to_otlp_kvarray(arena, field, &scope->n_attributes); if (field->via.map.size > 0 && scope->attributes == NULL) { return -1; @@ -974,6 +987,7 @@ static int add_msgpack_scope_fields( } static struct otlp_proto_logs_resource_state *append_logs_resource_state( + struct otlp_proto_arena *arena, Opentelemetry__Proto__Collector__Logs__V1__ExportLogsServiceRequest *export_logs, struct otlp_proto_logs_resource_state **states, size_t *state_count, @@ -989,11 +1003,11 @@ static struct otlp_proto_logs_resource_state *append_logs_resource_state( Opentelemetry__Proto__Logs__V1__ResourceLogs **tmp; msgpack_object *schema_url; - resource_log = flb_calloc(1, sizeof(Opentelemetry__Proto__Logs__V1__ResourceLogs)); - resource = flb_calloc(1, sizeof(Opentelemetry__Proto__Resource__V1__Resource)); + resource_log = otlp_proto_arena_calloc(arena, 1, + sizeof(Opentelemetry__Proto__Logs__V1__ResourceLogs)); + resource = otlp_proto_arena_calloc(arena, 1, + sizeof(Opentelemetry__Proto__Resource__V1__Resource)); if (resource_log == NULL || resource == NULL) { - flb_free(resource_log); - flb_free(resource); return NULL; } @@ -1001,20 +1015,16 @@ static struct otlp_proto_logs_resource_state *append_logs_resource_state( opentelemetry__proto__resource__v1__resource__init(resource); resource_log->resource = resource; - if (add_msgpack_attributes_to_resource(resource, resource_object) != 0) { - flb_free(resource); - flb_free(resource_log); + if (add_msgpack_attributes_to_resource(arena, resource, resource_object) != 0) { return NULL; } schema_url = resource_schema_url_object(resource_object, resource_body); if (schema_url != NULL && schema_url->type == MSGPACK_OBJECT_STR) { - resource_log->schema_url = flb_strndup(schema_url->via.str.ptr, - schema_url->via.str.size); + resource_log->schema_url = otlp_proto_arena_strndup(arena, + schema_url->via.str.ptr, + schema_url->via.str.size); if (resource_log->schema_url == NULL) { - otlp_kvarray_destroy(resource->attributes, resource->n_attributes); - flb_free(resource); - flb_free(resource_log); return NULL; } } @@ -1023,13 +1033,6 @@ static struct otlp_proto_logs_resource_state *append_logs_resource_state( sizeof(Opentelemetry__Proto__Logs__V1__ResourceLogs *) * (export_logs->n_resource_logs + 1)); if (tmp == NULL) { - otlp_kvarray_destroy(resource->attributes, resource->n_attributes); - if (resource_log->schema_url != NULL && - resource_log->schema_url != protobuf_c_empty_string) { - flb_free(resource_log->schema_url); - } - flb_free(resource); - flb_free(resource_log); return NULL; } @@ -1056,6 +1059,7 @@ static struct otlp_proto_logs_resource_state *append_logs_resource_state( } static struct otlp_proto_logs_scope_state *append_logs_scope_state( + struct otlp_proto_arena *arena, struct otlp_proto_logs_resource_state *resource_state, int64_t scope_id, uint64_t scope_hash, @@ -1068,11 +1072,11 @@ static struct otlp_proto_logs_scope_state *append_logs_scope_state( Opentelemetry__Proto__Logs__V1__ScopeLogs **tmp; msgpack_object *schema_url; - scope_log = flb_calloc(1, sizeof(Opentelemetry__Proto__Logs__V1__ScopeLogs)); - scope = flb_calloc(1, sizeof(Opentelemetry__Proto__Common__V1__InstrumentationScope)); + scope_log = otlp_proto_arena_calloc(arena, 1, + sizeof(Opentelemetry__Proto__Logs__V1__ScopeLogs)); + scope = otlp_proto_arena_calloc(arena, 1, + sizeof(Opentelemetry__Proto__Common__V1__InstrumentationScope)); if (scope_log == NULL || scope == NULL) { - flb_free(scope_log); - flb_free(scope); return NULL; } @@ -1080,26 +1084,17 @@ static struct otlp_proto_logs_scope_state *append_logs_scope_state( opentelemetry__proto__common__v1__instrumentation_scope__init(scope); scope_log->scope = scope; - if (add_msgpack_scope_fields(scope, scope_object) != 0) { - flb_free(scope->name); - flb_free(scope->version); - otlp_kvarray_destroy(scope->attributes, scope->n_attributes); - flb_free(scope); - flb_free(scope_log); + if (add_msgpack_scope_fields(arena, scope, scope_object) != 0) { return NULL; } if (scope_object != NULL && scope_object->type == MSGPACK_OBJECT_MAP) { schema_url = msgpack_map_get_object(&scope_object->via.map, "schema_url"); if (schema_url != NULL && schema_url->type == MSGPACK_OBJECT_STR) { - scope_log->schema_url = flb_strndup(schema_url->via.str.ptr, - schema_url->via.str.size); + scope_log->schema_url = otlp_proto_arena_strndup(arena, + schema_url->via.str.ptr, + schema_url->via.str.size); if (scope_log->schema_url == NULL) { - flb_free(scope->name); - flb_free(scope->version); - otlp_kvarray_destroy(scope->attributes, scope->n_attributes); - flb_free(scope); - flb_free(scope_log); return NULL; } } @@ -1109,15 +1104,6 @@ static struct otlp_proto_logs_scope_state *append_logs_scope_state( sizeof(Opentelemetry__Proto__Logs__V1__ScopeLogs *) * (resource_state->resource_log->n_scope_logs + 1)); if (tmp == NULL) { - if (scope_log->schema_url != NULL && - scope_log->schema_url != protobuf_c_empty_string) { - flb_free(scope_log->schema_url); - } - flb_free(scope->name); - flb_free(scope->version); - otlp_kvarray_destroy(scope->attributes, scope->n_attributes); - flb_free(scope); - flb_free(scope_log); return NULL; } @@ -1145,6 +1131,7 @@ static struct otlp_proto_logs_scope_state *append_logs_scope_state( } static int ensure_default_logs_scope_state( + struct otlp_proto_arena *arena, Opentelemetry__Proto__Collector__Logs__V1__ExportLogsServiceRequest *export_logs, struct otlp_proto_logs_resource_state **resource_states, size_t *resource_state_count, @@ -1162,7 +1149,8 @@ static int ensure_default_logs_scope_state( 0, resource_hash); if (*current_resource == NULL) { - *current_resource = append_logs_resource_state(export_logs, + *current_resource = append_logs_resource_state(arena, + export_logs, resource_states, resource_state_count, 0, @@ -1176,7 +1164,8 @@ static int ensure_default_logs_scope_state( *current_scope = find_logs_scope_state(*current_resource, 0, scope_hash); if (*current_scope == NULL) { - *current_scope = append_logs_scope_state(*current_resource, + *current_scope = append_logs_scope_state(arena, + *current_resource, 0, scope_hash, NULL); @@ -1188,7 +1177,8 @@ static int ensure_default_logs_scope_state( return 0; } -static int append_binary_id_field(ProtobufCBinaryData *field, +static int append_binary_id_field(struct otlp_proto_arena *arena, + ProtobufCBinaryData *field, msgpack_object *value, size_t expected_size) { @@ -1203,7 +1193,7 @@ static int append_binary_id_field(ProtobufCBinaryData *field, return 0; } - field->data = flb_malloc(value->via.bin.size); + field->data = otlp_proto_arena_alloc(arena, value->via.bin.size); if (field->data == NULL) { return -1; } @@ -1221,7 +1211,7 @@ static int append_binary_id_field(ProtobufCBinaryData *field, return 0; } - field->data = flb_calloc(1, expected_size); + field->data = otlp_proto_arena_calloc(arena, 1, expected_size); if (field->data == NULL) { return -1; } @@ -1232,7 +1222,6 @@ static int append_binary_id_field(ProtobufCBinaryData *field, char *str = (char *) value->via.str.ptr; if (!isxdigit(str[i * 2]) || !isxdigit(str[i * 2 + 1])) { - flb_free(field->data); field->data = NULL; return -1; } @@ -1251,7 +1240,8 @@ static int append_binary_id_field(ProtobufCBinaryData *field, return 0; } -static int log_record_to_proto(Opentelemetry__Proto__Logs__V1__LogRecord *record, +static int log_record_to_proto(struct otlp_proto_arena *arena, + Opentelemetry__Proto__Logs__V1__LogRecord *record, struct flb_log_event *event, const char **logs_body_keys, size_t logs_body_key_count, @@ -1311,8 +1301,9 @@ static int log_record_to_proto(Opentelemetry__Proto__Logs__V1__LogRecord *record field = msgpack_map_get_object(&otlp_metadata->via.map, "severity_text"); if (field != NULL && field->type == MSGPACK_OBJECT_STR) { - record->severity_text = flb_strndup(field->via.str.ptr, - field->via.str.size); + record->severity_text = otlp_proto_arena_strndup(arena, + field->via.str.ptr, + field->via.str.size); if (record->severity_text == NULL) { return -1; } @@ -1320,28 +1311,30 @@ static int log_record_to_proto(Opentelemetry__Proto__Logs__V1__LogRecord *record field = msgpack_map_get_object(&otlp_metadata->via.map, "attributes"); if (field != NULL && field->type == MSGPACK_OBJECT_MAP) { - attrs = msgpack_map_to_otlp_kvarray(field, &count); + attrs = msgpack_map_to_otlp_kvarray(arena, field, &count); if (field->via.map.size > 0 && attrs == NULL) { return -1; } - if (append_kvarrays(&record->attributes, + if (append_kvarrays(arena, + &record->attributes, &record->n_attributes, attrs, count) != 0) { - otlp_kvarray_destroy(attrs, count); return -1; } } - if (append_binary_id_field(&record->trace_id, + if (append_binary_id_field(arena, + &record->trace_id, msgpack_map_get_object(&otlp_metadata->via.map, "trace_id"), 16) != 0) { return -1; } - if (append_binary_id_field(&record->span_id, + if (append_binary_id_field(arena, + &record->span_id, msgpack_map_get_object(&otlp_metadata->via.map, "span_id"), 8) != 0) { @@ -1349,7 +1342,8 @@ static int log_record_to_proto(Opentelemetry__Proto__Logs__V1__LogRecord *record } } - if (log_record_set_body_and_attributes(record, + if (log_record_set_body_and_attributes(arena, + record, event, logs_body_keys, logs_body_key_count, @@ -1360,29 +1354,11 @@ static int log_record_to_proto(Opentelemetry__Proto__Logs__V1__LogRecord *record return 0; } -static void destroy_log_record(Opentelemetry__Proto__Logs__V1__LogRecord *record) -{ - if (record == NULL) { - return; - } - - otlp_any_value_destroy(record->body); - otlp_kvarray_destroy(record->attributes, record->n_attributes); - if (record->severity_text != NULL && - record->severity_text != protobuf_c_empty_string) { - flb_free(record->severity_text); - } - flb_free(record->span_id.data); - flb_free(record->trace_id.data); - flb_free(record); -} - -static void destroy_export_logs( +static void destroy_export_logs_arrays( Opentelemetry__Proto__Collector__Logs__V1__ExportLogsServiceRequest *export_logs) { size_t index; size_t inner; - size_t record_index; Opentelemetry__Proto__Logs__V1__ResourceLogs *resource_log; Opentelemetry__Proto__Logs__V1__ScopeLogs *scope_log; @@ -1402,44 +1378,10 @@ static void destroy_export_logs( continue; } - for (record_index = 0; - record_index < scope_log->n_log_records; - record_index++) { - destroy_log_record(scope_log->log_records[record_index]); - } - flb_free(scope_log->log_records); - if (scope_log->scope != NULL) { - if (scope_log->scope->name != NULL && - scope_log->scope->name != protobuf_c_empty_string) { - flb_free(scope_log->scope->name); - } - if (scope_log->scope->version != NULL && - scope_log->scope->version != protobuf_c_empty_string) { - flb_free(scope_log->scope->version); - } - otlp_kvarray_destroy(scope_log->scope->attributes, - scope_log->scope->n_attributes); - flb_free(scope_log->scope); - } - if (scope_log->schema_url != NULL && - scope_log->schema_url != protobuf_c_empty_string) { - flb_free(scope_log->schema_url); - } - flb_free(scope_log); } flb_free(resource_log->scope_logs); - if (resource_log->resource != NULL) { - otlp_kvarray_destroy(resource_log->resource->attributes, - resource_log->resource->n_attributes); - flb_free(resource_log->resource); - } - if (resource_log->schema_url != NULL && - resource_log->schema_url != protobuf_c_empty_string) { - flb_free(resource_log->schema_url); - } - flb_free(resource_log); } flb_free(export_logs->resource_logs); @@ -1598,12 +1540,14 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, flb_sds_t output; struct flb_log_event event; struct flb_log_event_decoder decoder; + struct otlp_proto_arena arena; msgpack_object *group_metadata; msgpack_object *group_body; msgpack_object *resource_object; msgpack_object *scope_object; uint64_t resource_hash; uint64_t scope_hash; + Opentelemetry__Proto__Logs__V1__LogRecord **tmp; struct otlp_proto_logs_scope_state *current_scope; struct otlp_proto_logs_resource_state *current_resource; struct otlp_proto_logs_resource_state *resource_states; @@ -1637,6 +1581,10 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, opentelemetry__proto__collector__logs__v1__export_logs_service_request__init( &export_logs); + if (otlp_proto_arena_init(&arena) != 0) { + set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_NOT_SUPPORTED, ENOMEM); + return NULL; + } current_scope = NULL; current_resource = NULL; @@ -1647,6 +1595,7 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, (char *) event_chunk_data, event_chunk_size); if (ret != FLB_EVENT_DECODER_SUCCESS) { + otlp_proto_arena_destroy(&arena); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_INVALID_ARGUMENT, EINVAL); return NULL; } @@ -1658,7 +1607,8 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, ret = flb_log_event_decoder_get_record_type(&event, &record_type); if (ret != 0) { flb_log_event_decoder_destroy(&decoder); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); destroy_logs_resource_states(resource_states, resource_state_count); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_INVALID_LOG_EVENT, EINVAL); return NULL; @@ -1677,7 +1627,8 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, msgpack_map_get_int64(&group_metadata->via.map, "scope_id", &scope_id) != 0) { if (require_otel_metadata == FLB_TRUE) { flb_log_event_decoder_destroy(&decoder); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); destroy_logs_resource_states(resource_states, resource_state_count); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_INVALID_LOG_EVENT, EINVAL); return NULL; @@ -1704,7 +1655,8 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, resource_id, resource_hash); if (current_resource == NULL) { - current_resource = append_logs_resource_state(&export_logs, + current_resource = append_logs_resource_state(&arena, + &export_logs, &resource_states, &resource_state_count, resource_id, @@ -1713,7 +1665,8 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, group_body); if (current_resource == NULL) { flb_log_event_decoder_destroy(&decoder); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); destroy_logs_resource_states(resource_states, resource_state_count); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_NOT_SUPPORTED, ENOMEM); return NULL; @@ -1724,13 +1677,15 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, scope_id, scope_hash); if (current_scope == NULL) { - current_scope = append_logs_scope_state(current_resource, + current_scope = append_logs_scope_state(&arena, + current_resource, scope_id, scope_hash, scope_object); if (current_scope == NULL) { flb_log_event_decoder_destroy(&decoder); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); destroy_logs_resource_states(resource_states, resource_state_count); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_NOT_SUPPORTED, ENOMEM); return NULL; @@ -1748,33 +1703,35 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, if (current_scope == NULL) { if (require_otel_metadata == FLB_TRUE) { flb_log_event_decoder_destroy(&decoder); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); destroy_logs_resource_states(resource_states, resource_state_count); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_INVALID_LOG_EVENT, EINVAL); return NULL; } - if (ensure_default_logs_scope_state(&export_logs, + if (ensure_default_logs_scope_state(&arena, + &export_logs, &resource_states, &resource_state_count, ¤t_resource, ¤t_scope) != 0) { flb_log_event_decoder_destroy(&decoder); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); destroy_logs_resource_states(resource_states, resource_state_count); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_NOT_SUPPORTED, ENOMEM); return NULL; } } - Opentelemetry__Proto__Logs__V1__LogRecord **tmp; - tmp = flb_realloc(current_scope->scope_log->log_records, sizeof(Opentelemetry__Proto__Logs__V1__LogRecord *) * (current_scope->scope_log->n_log_records + 1)); if (tmp == NULL) { flb_log_event_decoder_destroy(&decoder); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); destroy_logs_resource_states(resource_states, resource_state_count); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_NOT_SUPPORTED, ENOMEM); return NULL; @@ -1782,12 +1739,14 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, current_scope->scope_log->log_records = tmp; - record = flb_calloc(1, sizeof(Opentelemetry__Proto__Logs__V1__LogRecord)); + record = otlp_proto_arena_calloc(&arena, 1, + sizeof(Opentelemetry__Proto__Logs__V1__LogRecord)); current_scope->scope_log->log_records[ current_scope->scope_log->n_log_records] = record; if (record == NULL) { flb_log_event_decoder_destroy(&decoder); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); destroy_logs_resource_states(resource_states, resource_state_count); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_NOT_SUPPORTED, ENOMEM); return NULL; @@ -1796,16 +1755,17 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, opentelemetry__proto__logs__v1__log_record__init(record); if (log_record_to_proto( + &arena, record, &event, logs_body_keys, logs_body_key_count, logs_body_key_attributes) != 0) { - destroy_log_record(record); current_scope->scope_log->log_records[ current_scope->scope_log->n_log_records] = NULL; flb_log_event_decoder_destroy(&decoder); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); destroy_logs_resource_states(resource_states, resource_state_count); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_NOT_SUPPORTED, ENOMEM); return NULL; @@ -1819,7 +1779,8 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, if (ret != FLB_EVENT_DECODER_SUCCESS && ret != FLB_EVENT_DECODER_ERROR_INSUFFICIENT_DATA) { - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_INVALID_LOG_EVENT, EINVAL); return NULL; } @@ -1828,7 +1789,8 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, opentelemetry__proto__collector__logs__v1__export_logs_service_request__get_packed_size( &export_logs)); if (output == NULL) { - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); set_error(result, FLB_OPENTELEMETRY_OTLP_PROTO_NOT_SUPPORTED, ENOMEM); return NULL; } @@ -1837,7 +1799,8 @@ flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data, opentelemetry__proto__collector__logs__v1__export_logs_service_request__pack( &export_logs, (uint8_t *) output)); - destroy_export_logs(&export_logs); + destroy_export_logs_arrays(&export_logs); + otlp_proto_arena_destroy(&arena); set_result(result, FLB_OPENTELEMETRY_OTLP_PROTO_SUCCESS); return output; diff --git a/tests/internal/opentelemetry.c b/tests/internal/opentelemetry.c index 404cca8e1b9..13e9add6538 100644 --- a/tests/internal/opentelemetry.c +++ b/tests/internal/opentelemetry.c @@ -2474,6 +2474,7 @@ void test_opentelemetry_logs_otlp_proto_from_plain_logs() struct flb_log_event_encoder encoder; struct flb_opentelemetry_otlp_logs_options options; Opentelemetry__Proto__Collector__Logs__V1__ExportLogsServiceRequest *decoded; + char empty_binary[] = {(char) 0xc4, 0x00}; timestamp.tm.tv_sec = 1640995200; timestamp.tm.tv_nsec = 0; @@ -2499,7 +2500,10 @@ void test_opentelemetry_logs_otlp_proto_from_plain_logs() ret = flb_log_event_encoder_append_body_values( &encoder, FLB_LOG_EVENT_CSTRING_VALUE("message"), - FLB_LOG_EVENT_CSTRING_VALUE("hello from dummy")); + FLB_LOG_EVENT_CSTRING_VALUE("hello from dummy"), + FLB_LOG_EVENT_CSTRING_VALUE("empty_binary"), + FLB_LOG_EVENT_MSGPACK_RAW_VALUE(empty_binary, + sizeof(empty_binary))); TEST_CHECK(ret == FLB_EVENT_ENCODER_SUCCESS); if (ret != FLB_EVENT_ENCODER_SUCCESS) { flb_log_event_encoder_destroy(&encoder); @@ -2514,6 +2518,7 @@ void test_opentelemetry_logs_otlp_proto_from_plain_logs() memset(&options, 0, sizeof(options)); options.logs_require_otel_metadata = FLB_FALSE; + options.logs_body_key_attributes = FLB_TRUE; actual = flb_opentelemetry_logs_to_otlp_proto(encoder.output_buffer, encoder.output_length, @@ -2539,6 +2544,15 @@ void test_opentelemetry_logs_otlp_proto_from_plain_logs() OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_STRING_VALUE); TEST_CHECK(strcmp(decoded->resource_logs[0]->scope_logs[0]->log_records[0]->body->string_value, "hello from dummy") == 0); + TEST_CHECK(decoded->resource_logs[0]->scope_logs[0]->log_records[0]->n_attributes == 1); + TEST_CHECK(strcmp(decoded->resource_logs[0]->scope_logs[0]->log_records[0] + ->attributes[0]->key, + "empty_binary") == 0); + TEST_CHECK(decoded->resource_logs[0]->scope_logs[0]->log_records[0] + ->attributes[0]->value->value_case == + OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_BYTES_VALUE); + TEST_CHECK(decoded->resource_logs[0]->scope_logs[0]->log_records[0] + ->attributes[0]->value->bytes_value.len == 0); opentelemetry__proto__collector__logs__v1__export_logs_service_request__free_unpacked(decoded, NULL); }