diff --git a/include/cmetrics/cmt_encode_opentelemetry.h b/include/cmetrics/cmt_encode_opentelemetry.h index fd7c10a..98f2bbd 100644 --- a/include/cmetrics/cmt_encode_opentelemetry.h +++ b/include/cmetrics/cmt_encode_opentelemetry.h @@ -40,7 +40,36 @@ struct cmt_opentelemetry_context struct cmt *cmt; }; +struct cmt_opentelemetry_batch { + cfl_sds_t payload; + size_t data_point_count; +}; + +struct cmt_opentelemetry_batches { + size_t count; + struct cmt_opentelemetry_batch *entries; +}; + cfl_sds_t cmt_encode_opentelemetry_create(struct cmt *cmt); void cmt_encode_opentelemetry_destroy(cfl_sds_t text); +/* + * Split an encoded OTLP ExportMetricsServiceRequest without changing metric, + * resource, or scope metadata. A zero limit returns the original request as a + * single batch. The returned payloads are owned by the batch collection. + */ +struct cmt_opentelemetry_batches * +cmt_encode_opentelemetry_split_payload(const void *payload, + size_t payload_size, + size_t max_data_points, + int *result); + +struct cmt_opentelemetry_batches * +cmt_encode_opentelemetry_create_batches(struct cmt *context, + size_t max_data_points, + int *result); + +void cmt_encode_opentelemetry_destroy_batches( + struct cmt_opentelemetry_batches *batches); + #endif diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index b79dcb3..a4bed29 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -26,6 +26,7 @@ set(src cmt_filter.c cmetrics.c cmt_encode_opentelemetry.c + cmt_encode_opentelemetry_batch.c cmt_decode_opentelemetry.c cmt_encode_prometheus.c cmt_encode_prometheus_remote_write.c diff --git a/src/cmt_encode_opentelemetry_batch.c b/src/cmt_encode_opentelemetry_batch.c new file mode 100644 index 0000000..e4f4d59 --- /dev/null +++ b/src/cmt_encode_opentelemetry_batch.c @@ -0,0 +1,805 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* CMetrics + * ======== + * Copyright 2026 The CMetrics Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include +#include +#include + +#include + +struct metrics_batch_view { + size_t data_point_count; + Opentelemetry__Proto__Collector__Metrics__V1__ExportMetricsServiceRequest request; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics *active_resource; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics *active_scope; + const Opentelemetry__Proto__Metrics__V1__ResourceMetrics *source_resource; + const Opentelemetry__Proto__Metrics__V1__ScopeMetrics *source_scope; +}; + +static void set_result(int *result, int value) +{ + if (result != NULL) { + *result = value; + } +} + +static void metrics_batch_view_init(struct metrics_batch_view *batch) +{ + memset(batch, 0, sizeof(struct metrics_batch_view)); + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__init( + &batch->request); +} + +static void metric_view_destroy(Opentelemetry__Proto__Metrics__V1__Metric *metric) +{ + if (metric == NULL) { + return; + } + + if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_GAUGE) { + free(metric->gauge); + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUM) { + free(metric->sum); + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_HISTOGRAM) { + free(metric->histogram); + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_EXPONENTIAL_HISTOGRAM) { + free(metric->exponential_histogram); + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUMMARY) { + free(metric->summary); + } + + free(metric); +} + +static void metrics_batch_view_destroy(struct metrics_batch_view *batch) +{ + size_t resource_index; + size_t scope_index; + size_t metric_index; + Opentelemetry__Proto__Metrics__V1__Metric *metric; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics *scope; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics *resource; + + for (resource_index = 0; + resource_index < batch->request.n_resource_metrics; + resource_index++) { + resource = batch->request.resource_metrics[resource_index]; + + for (scope_index = 0; + scope_index < resource->n_scope_metrics; + scope_index++) { + scope = resource->scope_metrics[scope_index]; + + for (metric_index = 0; + metric_index < scope->n_metrics; + metric_index++) { + metric = scope->metrics[metric_index]; + metric_view_destroy(metric); + } + + free(scope->metrics); + free(scope); + } + + free(resource->scope_metrics); + free(resource); + } + + free(batch->request.resource_metrics); + metrics_batch_view_init(batch); +} + +void cmt_encode_opentelemetry_destroy_batches( + struct cmt_opentelemetry_batches *batches) +{ + size_t index; + + if (batches == NULL) { + return; + } + + for (index = 0; index < batches->count; index++) { + cfl_sds_destroy(batches->entries[index].payload); + } + + free(batches->entries); + free(batches); +} + +static int batches_append(struct cmt_opentelemetry_batches *batches, + cfl_sds_t payload, + size_t data_point_count) +{ + size_t count; + struct cmt_opentelemetry_batch *entries; + + count = batches->count; + if (count >= SIZE_MAX / sizeof(struct cmt_opentelemetry_batch)) { + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + entries = realloc( + batches->entries, + (count + 1) * sizeof(struct cmt_opentelemetry_batch)); + if (entries == NULL) { + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + entries[count].payload = payload; + entries[count].data_point_count = data_point_count; + batches->entries = entries; + batches->count++; + + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; +} + +static int batch_add_resource( + struct metrics_batch_view *batch, + const Opentelemetry__Proto__Metrics__V1__ResourceMetrics *source) +{ + size_t count; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics *resource; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics **resources; + + resource = calloc(1, sizeof(Opentelemetry__Proto__Metrics__V1__ResourceMetrics)); + if (resource == NULL) { + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + *resource = *source; + resource->n_scope_metrics = 0; + resource->scope_metrics = NULL; + + count = batch->request.n_resource_metrics; + if (count >= SIZE_MAX / sizeof(Opentelemetry__Proto__Metrics__V1__ResourceMetrics *)) { + free(resource); + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + resources = realloc( + batch->request.resource_metrics, + (count + 1) * + sizeof(Opentelemetry__Proto__Metrics__V1__ResourceMetrics *)); + if (resources == NULL) { + free(resource); + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + resources[count] = resource; + batch->request.resource_metrics = resources; + batch->request.n_resource_metrics++; + batch->active_resource = resource; + batch->active_scope = NULL; + batch->source_resource = source; + batch->source_scope = NULL; + + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; +} + +static int batch_ensure_resource( + struct metrics_batch_view *batch, + const Opentelemetry__Proto__Metrics__V1__ResourceMetrics *source) +{ + if (batch->source_resource == source && batch->active_resource != NULL) { + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; + } + + return batch_add_resource(batch, source); +} + +static int batch_add_scope( + struct metrics_batch_view *batch, + const Opentelemetry__Proto__Metrics__V1__ScopeMetrics *source) +{ + size_t count; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics *scope; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics **scopes; + + scope = calloc(1, sizeof(Opentelemetry__Proto__Metrics__V1__ScopeMetrics)); + if (scope == NULL) { + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + *scope = *source; + scope->n_metrics = 0; + scope->metrics = NULL; + + count = batch->active_resource->n_scope_metrics; + if (count >= SIZE_MAX / sizeof(Opentelemetry__Proto__Metrics__V1__ScopeMetrics *)) { + free(scope); + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + scopes = realloc( + batch->active_resource->scope_metrics, + (count + 1) * sizeof(Opentelemetry__Proto__Metrics__V1__ScopeMetrics *)); + if (scopes == NULL) { + free(scope); + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + scopes[count] = scope; + batch->active_resource->scope_metrics = scopes; + batch->active_resource->n_scope_metrics++; + batch->active_scope = scope; + batch->source_scope = source; + + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; +} + +static int batch_ensure_scope( + struct metrics_batch_view *batch, + const Opentelemetry__Proto__Metrics__V1__ResourceMetrics *source_resource, + const Opentelemetry__Proto__Metrics__V1__ScopeMetrics *source_scope) +{ + int result; + + result = batch_ensure_resource(batch, source_resource); + if (result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + return result; + } + + if (batch->source_scope == source_scope && batch->active_scope != NULL) { + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; + } + + return batch_add_scope(batch, source_scope); +} + +static int metric_data_point_count( + const Opentelemetry__Proto__Metrics__V1__Metric *metric, + size_t *count) +{ + *count = 0; + + if (metric == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + + if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA__NOT_SET) { + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_GAUGE) { + if (metric->gauge == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + *count = metric->gauge->n_data_points; + if (*count > 0 && metric->gauge->data_points == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUM) { + if (metric->sum == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + *count = metric->sum->n_data_points; + if (*count > 0 && metric->sum->data_points == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_HISTOGRAM) { + if (metric->histogram == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + *count = metric->histogram->n_data_points; + if (*count > 0 && metric->histogram->data_points == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_EXPONENTIAL_HISTOGRAM) { + if (metric->exponential_histogram == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + *count = metric->exponential_histogram->n_data_points; + if (*count > 0 && metric->exponential_histogram->data_points == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUMMARY) { + if (metric->summary == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + *count = metric->summary->n_data_points; + if (*count > 0 && metric->summary->data_points == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + } + + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; +} + +static Opentelemetry__Proto__Metrics__V1__Metric *metric_view_create( + const Opentelemetry__Proto__Metrics__V1__Metric *source, + size_t offset, + size_t count) +{ + Opentelemetry__Proto__Metrics__V1__Metric *metric; + Opentelemetry__Proto__Metrics__V1__Gauge *gauge; + Opentelemetry__Proto__Metrics__V1__Sum *sum; + Opentelemetry__Proto__Metrics__V1__Histogram *histogram; + Opentelemetry__Proto__Metrics__V1__ExponentialHistogram *exp_histogram; + Opentelemetry__Proto__Metrics__V1__Summary *summary; + + metric = calloc(1, sizeof(Opentelemetry__Proto__Metrics__V1__Metric)); + if (metric == NULL) { + return NULL; + } + + *metric = *source; + + if (source->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_GAUGE) { + gauge = calloc(1, sizeof(Opentelemetry__Proto__Metrics__V1__Gauge)); + if (gauge == NULL) { + free(metric); + return NULL; + } + *gauge = *source->gauge; + gauge->n_data_points = count; + gauge->data_points = count > 0 ? source->gauge->data_points + offset : NULL; + metric->gauge = gauge; + } + else if (source->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUM) { + sum = calloc(1, sizeof(Opentelemetry__Proto__Metrics__V1__Sum)); + if (sum == NULL) { + free(metric); + return NULL; + } + *sum = *source->sum; + sum->n_data_points = count; + sum->data_points = count > 0 ? source->sum->data_points + offset : NULL; + metric->sum = sum; + } + else if (source->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_HISTOGRAM) { + histogram = calloc(1, sizeof(Opentelemetry__Proto__Metrics__V1__Histogram)); + if (histogram == NULL) { + free(metric); + return NULL; + } + *histogram = *source->histogram; + histogram->n_data_points = count; + histogram->data_points = count > 0 ? source->histogram->data_points + offset : NULL; + metric->histogram = histogram; + } + else if (source->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_EXPONENTIAL_HISTOGRAM) { + exp_histogram = calloc( + 1, + sizeof(Opentelemetry__Proto__Metrics__V1__ExponentialHistogram)); + if (exp_histogram == NULL) { + free(metric); + return NULL; + } + *exp_histogram = *source->exponential_histogram; + exp_histogram->n_data_points = count; + exp_histogram->data_points = count > 0 ? + source->exponential_histogram->data_points + offset : + NULL; + metric->exponential_histogram = exp_histogram; + } + else if (source->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUMMARY) { + summary = calloc(1, sizeof(Opentelemetry__Proto__Metrics__V1__Summary)); + if (summary == NULL) { + free(metric); + return NULL; + } + *summary = *source->summary; + summary->n_data_points = count; + summary->data_points = count > 0 ? source->summary->data_points + offset : NULL; + metric->summary = summary; + } + + return metric; +} + +static int batch_add_metric( + struct metrics_batch_view *batch, + const Opentelemetry__Proto__Metrics__V1__ResourceMetrics *source_resource, + const Opentelemetry__Proto__Metrics__V1__ScopeMetrics *source_scope, + const Opentelemetry__Proto__Metrics__V1__Metric *source_metric, + size_t offset, + size_t count) +{ + int result; + size_t metric_count; + Opentelemetry__Proto__Metrics__V1__Metric *metric; + Opentelemetry__Proto__Metrics__V1__Metric **metrics; + + result = batch_ensure_scope(batch, source_resource, source_scope); + if (result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + return result; + } + + metric = metric_view_create(source_metric, offset, count); + if (metric == NULL) { + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + metric_count = batch->active_scope->n_metrics; + if (metric_count >= SIZE_MAX / sizeof(Opentelemetry__Proto__Metrics__V1__Metric *)) { + metric_view_destroy(metric); + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + metrics = realloc( + batch->active_scope->metrics, + (metric_count + 1) * sizeof(Opentelemetry__Proto__Metrics__V1__Metric *)); + if (metrics == NULL) { + metric_view_destroy(metric); + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + metrics[metric_count] = metric; + batch->active_scope->metrics = metrics; + batch->active_scope->n_metrics++; + batch->data_point_count += count; + + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; +} + +static int batch_pack(struct metrics_batch_view *batch, + struct cmt_opentelemetry_batches *batches) +{ + int result; + size_t payload_size; + size_t packed_size; + cfl_sds_t payload; + + if (batch->request.n_resource_metrics == 0) { + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; + } + + payload_size = + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__get_packed_size( + &batch->request); + payload = cfl_sds_create_size(payload_size); + if (payload == NULL) { + metrics_batch_view_destroy(batch); + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + packed_size = + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__pack( + &batch->request, + (uint8_t *) payload); + if (packed_size != payload_size) { + cfl_sds_destroy(payload); + metrics_batch_view_destroy(batch); + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + + cfl_sds_len_set(payload, packed_size); + result = batches_append(batches, payload, batch->data_point_count); + if (result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + cfl_sds_destroy(payload); + } + + metrics_batch_view_destroy(batch); + + return result; +} + +static int request_data_point_count( + Opentelemetry__Proto__Collector__Metrics__V1__ExportMetricsServiceRequest *request, + size_t *total) +{ + int result; + size_t resource_index; + size_t scope_index; + size_t metric_index; + size_t count; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics *scope; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics *resource; + + *total = 0; + + for (resource_index = 0; resource_index < request->n_resource_metrics; resource_index++) { + resource = request->resource_metrics[resource_index]; + if (resource == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + + for (scope_index = 0; scope_index < resource->n_scope_metrics; scope_index++) { + scope = resource->scope_metrics[scope_index]; + if (scope == NULL) { + return CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR; + } + + for (metric_index = 0; metric_index < scope->n_metrics; metric_index++) { + result = metric_data_point_count(scope->metrics[metric_index], &count); + if (result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + return result; + } + if (count > SIZE_MAX - *total) { + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + *total += count; + } + } + } + + return CMT_ENCODE_OPENTELEMETRY_SUCCESS; +} + +static int append_original_payload( + struct cmt_opentelemetry_batches *batches, + const void *payload, + size_t payload_size, + size_t data_point_count) +{ + int result; + cfl_sds_t copy; + + copy = cfl_sds_create_size(payload_size); + if (copy == NULL) { + return CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR; + } + + memcpy(copy, payload, payload_size); + cfl_sds_len_set(copy, payload_size); + result = batches_append(batches, copy, data_point_count); + if (result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + cfl_sds_destroy(copy); + } + + return result; +} + +struct cmt_opentelemetry_batches * +cmt_encode_opentelemetry_split_payload(const void *payload, + size_t payload_size, + size_t max_data_points, + int *result) +{ + int local_result; + size_t resource_index; + size_t scope_index; + size_t metric_index; + size_t data_point_count; + size_t offset; + size_t remaining; + size_t batch_count; + size_t total_data_points; + Opentelemetry__Proto__Metrics__V1__Metric *metric; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics *scope; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics *resource; + Opentelemetry__Proto__Collector__Metrics__V1__ExportMetricsServiceRequest *request; + struct cmt_opentelemetry_batches *batches; + struct metrics_batch_view batch; + + if (payload == NULL) { + errno = EINVAL; + set_result(result, CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR); + return NULL; + } + + batches = calloc(1, sizeof(struct cmt_opentelemetry_batches)); + if (batches == NULL) { + errno = ENOMEM; + set_result(result, CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR); + return NULL; + } + + request = + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__unpack( + NULL, + payload_size, + (const uint8_t *) payload); + if (request == NULL) { + errno = EINVAL; + set_result(result, CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR); + cmt_encode_opentelemetry_destroy_batches(batches); + return NULL; + } + + local_result = request_data_point_count(request, &total_data_points); + if (local_result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + goto error; + } + + if (max_data_points == 0 || total_data_points <= max_data_points) { + local_result = append_original_payload(batches, + payload, + payload_size, + total_data_points); + if (local_result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + goto error; + } + + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__free_unpacked( + request, + NULL); + set_result(result, CMT_ENCODE_OPENTELEMETRY_SUCCESS); + return batches; + } + + local_result = CMT_ENCODE_OPENTELEMETRY_SUCCESS; + metrics_batch_view_init(&batch); + + for (resource_index = 0; + resource_index < request->n_resource_metrics && + local_result == CMT_ENCODE_OPENTELEMETRY_SUCCESS; + resource_index++) { + resource = request->resource_metrics[resource_index]; + + if (resource->n_scope_metrics == 0) { + local_result = batch_ensure_resource(&batch, resource); + continue; + } + + for (scope_index = 0; + scope_index < resource->n_scope_metrics && + local_result == CMT_ENCODE_OPENTELEMETRY_SUCCESS; + scope_index++) { + scope = resource->scope_metrics[scope_index]; + + if (scope->n_metrics == 0) { + local_result = batch_ensure_scope(&batch, resource, scope); + continue; + } + + for (metric_index = 0; + metric_index < scope->n_metrics && + local_result == CMT_ENCODE_OPENTELEMETRY_SUCCESS; + metric_index++) { + metric = scope->metrics[metric_index]; + local_result = metric_data_point_count(metric, &data_point_count); + if (local_result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + break; + } + + if (data_point_count == 0) { + local_result = batch_add_metric(&batch, + resource, + scope, + metric, + 0, + 0); + continue; + } + + offset = 0; + while (offset < data_point_count && + local_result == CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + if (batch.data_point_count == max_data_points) { + local_result = batch_pack(&batch, batches); + if (local_result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + break; + } + } + + remaining = data_point_count - offset; + batch_count = max_data_points - batch.data_point_count; + if (batch_count > remaining) { + batch_count = remaining; + } + + local_result = batch_add_metric(&batch, + resource, + scope, + metric, + offset, + batch_count); + offset += batch_count; + } + } + } + } + + if (local_result == CMT_ENCODE_OPENTELEMETRY_SUCCESS && + batch.request.n_resource_metrics > 0) { + local_result = batch_pack(&batch, batches); + } + else { + metrics_batch_view_destroy(&batch); + } + + if (local_result != CMT_ENCODE_OPENTELEMETRY_SUCCESS) { + goto error; + } + + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__free_unpacked( + request, + NULL); + set_result(result, CMT_ENCODE_OPENTELEMETRY_SUCCESS); + return batches; + +error: + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__free_unpacked( + request, + NULL); + cmt_encode_opentelemetry_destroy_batches(batches); + if (local_result == CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR) { + errno = ENOMEM; + } + else { + errno = EINVAL; + } + set_result(result, local_result); + return NULL; +} + +static struct cmt_opentelemetry_batches *empty_batches_create(int *result) +{ + struct cmt_opentelemetry_batches *batches; + + batches = calloc(1, sizeof(struct cmt_opentelemetry_batches)); + if (batches == NULL) { + errno = ENOMEM; + set_result(result, CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR); + return NULL; + } + + set_result(result, CMT_ENCODE_OPENTELEMETRY_SUCCESS); + return batches; +} + +struct cmt_opentelemetry_batches * +cmt_encode_opentelemetry_create_batches(struct cmt *context, + size_t max_data_points, + int *result) +{ + cfl_sds_t payload; + struct cmt_opentelemetry_batches *batches; + + if (context == NULL) { + errno = EINVAL; + set_result(result, CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR); + return NULL; + } + + payload = cmt_encode_opentelemetry_create(context); + if (payload == NULL) { + errno = ENOMEM; + set_result(result, CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR); + return NULL; + } + + if (cfl_sds_len(payload) == 0) { + cmt_encode_opentelemetry_destroy(payload); + return empty_batches_create(result); + } + + batches = cmt_encode_opentelemetry_split_payload(payload, + cfl_sds_len(payload), + max_data_points, + result); + cmt_encode_opentelemetry_destroy(payload); + + return batches; +} diff --git a/tests/opentelemetry.c b/tests/opentelemetry.c index 918f442..c1f5b53 100644 --- a/tests/opentelemetry.c +++ b/tests/opentelemetry.c @@ -167,6 +167,26 @@ static int compare_text_lines(const void *a, const void *b) return strcmp(line_a, line_b); } +static char *duplicate_test_string(const char *source) +{ +#ifdef _WIN32 + return _strdup(source); +#else + return strdup(source); +#endif +} + +static char *tokenize_test_string(char *source, + const char *delimiters, + char **context) +{ +#ifdef _WIN32 + return strtok_s(source, delimiters, context); +#else + return strtok_r(source, delimiters, context); +#endif +} + static int are_texts_equivalent_ignoring_line_order(const char *left, const char *right) { char *left_copy; @@ -184,8 +204,8 @@ static int are_texts_equivalent_ignoring_line_order(const char *left, const char return CMT_FALSE; } - left_copy = strdup(left); - right_copy = strdup(right); + left_copy = duplicate_test_string(left); + right_copy = duplicate_test_string(right); if (left_copy == NULL || right_copy == NULL) { free(left_copy); free(right_copy); @@ -216,18 +236,18 @@ static int are_texts_equivalent_ignoring_line_order(const char *left, const char left_count = 0; saveptr = NULL; - line = strtok_r(left_copy, "\n", &saveptr); + line = tokenize_test_string(left_copy, "\n", &saveptr); while (line != NULL) { left_lines[left_count++] = line; - line = strtok_r(NULL, "\n", &saveptr); + line = tokenize_test_string(NULL, "\n", &saveptr); } right_count = 0; saveptr = NULL; - line = strtok_r(right_copy, "\n", &saveptr); + line = tokenize_test_string(right_copy, "\n", &saveptr); while (line != NULL) { right_lines[right_count++] = line; - line = strtok_r(NULL, "\n", &saveptr); + line = tokenize_test_string(NULL, "\n", &saveptr); } if (left_count != right_count) { @@ -1993,6 +2013,569 @@ static void test_opentelemetry_omitted_null_key_label_encoded(void) cmt_destroy(cmt); } +static void test_opentelemetry_data_point_batches(void) +{ + int index; + int result; + int ret; + int seen[11]; + size_t batch_index; + size_t resource_index; + size_t scope_index; + size_t metric_index; + size_t point_index; + size_t total_data_points; + uint64_t timestamp; + cfl_sds_t payload; + char *label_keys[] = {"series"}; + char *label_values[1]; + char *series[] = { + "series-0", "series-1", "series-2", "series-3", "series-4", "series-5", + "series-6", "series-7", "series-8", "series-9", "series-10" + }; + struct cmt *context; + struct cmt_gauge *gauge; + struct cmt_opentelemetry_batches *batches; + Opentelemetry__Proto__Metrics__V1__Metric *metric; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics *scope; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics *resource; + Opentelemetry__Proto__Metrics__V1__NumberDataPoint *point; + Opentelemetry__Proto__Collector__Metrics__V1__ExportMetricsServiceRequest *decoded; + + memset(seen, 0, sizeof(seen)); + context = cmt_create(); + TEST_CHECK(context != NULL); + if (context == NULL) { + return; + } + + gauge = cmt_gauge_create(context, + "test", + "batch", + "value", + "batching test", + 1, + label_keys); + TEST_CHECK(gauge != NULL); + if (gauge == NULL) { + cmt_destroy(context); + return; + } + + for (index = 0; index < 11; index++) { + label_values[0] = series[index]; + ret = cmt_gauge_set(gauge, + (uint64_t) index + 1, + (double) index, + 1, + label_values); + TEST_CHECK(ret == 0); + } + + batches = cmt_encode_opentelemetry_create_batches(context, 4, &result); + TEST_CHECK(result == CMT_ENCODE_OPENTELEMETRY_SUCCESS); + TEST_CHECK(batches != NULL); + if (batches == NULL) { + cmt_destroy(context); + return; + } + + TEST_CHECK(batches->count == 3); + TEST_CHECK(batches->entries[0].data_point_count == 4); + TEST_CHECK(batches->entries[1].data_point_count == 4); + TEST_CHECK(batches->entries[2].data_point_count == 3); + + total_data_points = 0; + for (batch_index = 0; batch_index < batches->count; batch_index++) { + decoded = + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__unpack( + NULL, + cfl_sds_len(batches->entries[batch_index].payload), + (uint8_t *) batches->entries[batch_index].payload); + TEST_CHECK(decoded != NULL); + if (decoded == NULL) { + continue; + } + + for (resource_index = 0; + resource_index < decoded->n_resource_metrics; + resource_index++) { + resource = decoded->resource_metrics[resource_index]; + for (scope_index = 0; scope_index < resource->n_scope_metrics; scope_index++) { + scope = resource->scope_metrics[scope_index]; + for (metric_index = 0; metric_index < scope->n_metrics; metric_index++) { + metric = scope->metrics[metric_index]; + TEST_CHECK(metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_GAUGE); + if (metric->data_case != + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_GAUGE) { + continue; + } + + TEST_CHECK(strcmp(metric->name, "test_batch_value") == 0); + TEST_CHECK(strcmp(metric->description, "batching test") == 0); + for (point_index = 0; + point_index < metric->gauge->n_data_points; + point_index++) { + point = metric->gauge->data_points[point_index]; + timestamp = point->time_unix_nano; + TEST_CHECK(timestamp >= 1 && timestamp <= 11); + if (timestamp >= 1 && timestamp <= 11) { + seen[timestamp - 1]++; + } + total_data_points++; + } + } + } + } + + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__free_unpacked( + decoded, + NULL); + } + + TEST_CHECK(total_data_points == 11); + for (index = 0; index < 11; index++) { + TEST_CHECK(seen[index] == 1); + } + + cmt_encode_opentelemetry_destroy_batches(batches); + + batches = cmt_encode_opentelemetry_create_batches(context, 11, &result); + TEST_CHECK(result == CMT_ENCODE_OPENTELEMETRY_SUCCESS); + TEST_CHECK(batches != NULL); + if (batches != NULL) { + TEST_CHECK(batches->count == 1); + TEST_CHECK(batches->entries[0].data_point_count == 11); + cmt_encode_opentelemetry_destroy_batches(batches); + } + + payload = cmt_encode_opentelemetry_create(context); + TEST_CHECK(payload != NULL); + if (payload == NULL) { + cmt_destroy(context); + return; + } + + batches = cmt_encode_opentelemetry_split_payload(payload, + cfl_sds_len(payload), + 0, + &result); + TEST_CHECK(result == CMT_ENCODE_OPENTELEMETRY_SUCCESS); + TEST_CHECK(batches != NULL); + if (batches != NULL) { + TEST_CHECK(batches->count == 1); + TEST_CHECK(batches->entries[0].data_point_count == 11); + TEST_CHECK(cfl_sds_len(batches->entries[0].payload) == cfl_sds_len(payload)); + TEST_CHECK(memcmp(batches->entries[0].payload, + payload, + cfl_sds_len(payload)) == 0); + cmt_encode_opentelemetry_destroy_batches(batches); + } + cmt_encode_opentelemetry_destroy(payload); + + cmt_destroy(context); + + batches = cmt_encode_opentelemetry_create_batches(NULL, 4, &result); + TEST_CHECK(batches == NULL); + TEST_CHECK(result == CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR); + + batches = cmt_encode_opentelemetry_split_payload(NULL, 0, 4, &result); + TEST_CHECK(batches == NULL); + TEST_CHECK(result == CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR); + + batches = cmt_encode_opentelemetry_split_payload("invalid", 7, 4, &result); + TEST_CHECK(batches == NULL); + TEST_CHECK(result == CMT_ENCODE_OPENTELEMETRY_INVALID_ARGUMENT_ERROR); +} + +static void test_opentelemetry_batches_empty_context(void) +{ + int result; + const char empty_payload = '\0'; + struct cmt *context; + struct cmt_opentelemetry_batches *batches; + + batches = cmt_encode_opentelemetry_split_payload(&empty_payload, + 0, + 0, + &result); + TEST_CHECK(result == CMT_ENCODE_OPENTELEMETRY_SUCCESS); + TEST_CHECK(batches != NULL); + if (batches != NULL) { + TEST_CHECK(batches->count == 1); + if (batches->count == 1) { + TEST_CHECK(batches->entries[0].data_point_count == 0); + TEST_CHECK(cfl_sds_len(batches->entries[0].payload) == 0); + } + cmt_encode_opentelemetry_destroy_batches(batches); + } + + context = cmt_create(); + TEST_CHECK(context != NULL); + if (context == NULL) { + return; + } + + batches = cmt_encode_opentelemetry_create_batches(context, 4, &result); + TEST_CHECK(result == CMT_ENCODE_OPENTELEMETRY_SUCCESS); + TEST_CHECK(batches != NULL); + if (batches != NULL) { + TEST_CHECK(batches->count <= 1); + if (batches->count == 1) { + TEST_CHECK(batches->entries[0].data_point_count == 0); + } + cmt_encode_opentelemetry_destroy_batches(batches); + } + + cmt_destroy(context); +} + +static void test_opentelemetry_batches_all_metric_types(void) +{ + int result; + int type_index; + int type_seen[5]; + int point_seen[5][3]; + size_t batch_index; + size_t metric_index; + size_t point_index; + size_t data_point_index; + size_t payload_size; + size_t total_data_points; + uint64_t timestamp; + cfl_sds_t payload; + const char *metric_names[] = { + "gauge", "sum", "histogram", "exponential_histogram", "summary" + }; + Opentelemetry__Proto__Metrics__V1__Metric metrics[5]; + Opentelemetry__Proto__Metrics__V1__Metric *metric_entries[5]; + Opentelemetry__Proto__Metrics__V1__Gauge gauge; + Opentelemetry__Proto__Metrics__V1__Sum sum; + Opentelemetry__Proto__Metrics__V1__Histogram histogram; + Opentelemetry__Proto__Metrics__V1__ExponentialHistogram exp_histogram; + Opentelemetry__Proto__Metrics__V1__Summary summary; + Opentelemetry__Proto__Metrics__V1__NumberDataPoint gauge_point_values[3]; + Opentelemetry__Proto__Metrics__V1__NumberDataPoint sum_point_values[3]; + Opentelemetry__Proto__Metrics__V1__NumberDataPoint *gauge_points[3]; + Opentelemetry__Proto__Metrics__V1__NumberDataPoint *sum_points[3]; + Opentelemetry__Proto__Metrics__V1__HistogramDataPoint histogram_point_values[3]; + Opentelemetry__Proto__Metrics__V1__HistogramDataPoint *histogram_points[3]; + Opentelemetry__Proto__Metrics__V1__ExponentialHistogramDataPoint + exp_histogram_point_values[3]; + Opentelemetry__Proto__Metrics__V1__ExponentialHistogramDataPoint + *exp_histogram_points[3]; + Opentelemetry__Proto__Metrics__V1__SummaryDataPoint summary_point_values[3]; + Opentelemetry__Proto__Metrics__V1__SummaryDataPoint *summary_points[3]; + Opentelemetry__Proto__Resource__V1__Resource resource_metadata; + Opentelemetry__Proto__Common__V1__InstrumentationScope scope_metadata; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics scope; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics *scopes[1]; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics resource; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics *resources[1]; + Opentelemetry__Proto__Metrics__V1__Metric *metric; + Opentelemetry__Proto__Metrics__V1__ScopeMetrics *decoded_scope; + Opentelemetry__Proto__Metrics__V1__ResourceMetrics *decoded_resource; + Opentelemetry__Proto__Collector__Metrics__V1__ExportMetricsServiceRequest request; + Opentelemetry__Proto__Collector__Metrics__V1__ExportMetricsServiceRequest *decoded; + struct cmt_opentelemetry_batches *batches; + + memset(type_seen, 0, sizeof(type_seen)); + memset(point_seen, 0, sizeof(point_seen)); + + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__init( + &request); + opentelemetry__proto__metrics__v1__resource_metrics__init(&resource); + opentelemetry__proto__metrics__v1__scope_metrics__init(&scope); + opentelemetry__proto__metrics__v1__gauge__init(&gauge); + opentelemetry__proto__metrics__v1__sum__init(&sum); + opentelemetry__proto__metrics__v1__histogram__init(&histogram); + opentelemetry__proto__metrics__v1__exponential_histogram__init(&exp_histogram); + opentelemetry__proto__metrics__v1__summary__init(&summary); + opentelemetry__proto__resource__v1__resource__init(&resource_metadata); + opentelemetry__proto__common__v1__instrumentation_scope__init(&scope_metadata); + + for (metric_index = 0; metric_index < 5; metric_index++) { + opentelemetry__proto__metrics__v1__metric__init(&metrics[metric_index]); + metric_entries[metric_index] = &metrics[metric_index]; + metrics[metric_index].name = (char *) metric_names[metric_index]; + metrics[metric_index].description = "batch preservation test"; + metrics[metric_index].unit = "1"; + } + + for (point_index = 0; point_index < 3; point_index++) { + opentelemetry__proto__metrics__v1__number_data_point__init( + &gauge_point_values[point_index]); + opentelemetry__proto__metrics__v1__number_data_point__init( + &sum_point_values[point_index]); + opentelemetry__proto__metrics__v1__histogram_data_point__init( + &histogram_point_values[point_index]); + opentelemetry__proto__metrics__v1__exponential_histogram_data_point__init( + &exp_histogram_point_values[point_index]); + opentelemetry__proto__metrics__v1__summary_data_point__init( + &summary_point_values[point_index]); + + gauge_points[point_index] = &gauge_point_values[point_index]; + gauge_point_values[point_index].time_unix_nano = 1000 + point_index; + gauge_point_values[point_index].value_case = + OPENTELEMETRY__PROTO__METRICS__V1__NUMBER_DATA_POINT__VALUE_AS_INT; + gauge_point_values[point_index].as_int = 10 + point_index; + + sum_points[point_index] = &sum_point_values[point_index]; + sum_point_values[point_index].time_unix_nano = 2000 + point_index; + sum_point_values[point_index].value_case = + OPENTELEMETRY__PROTO__METRICS__V1__NUMBER_DATA_POINT__VALUE_AS_DOUBLE; + sum_point_values[point_index].as_double = 20.5 + point_index; + + histogram_points[point_index] = &histogram_point_values[point_index]; + histogram_point_values[point_index].time_unix_nano = 3000 + point_index; + histogram_point_values[point_index].count = 30 + point_index; + + exp_histogram_points[point_index] = &exp_histogram_point_values[point_index]; + exp_histogram_point_values[point_index].time_unix_nano = 4000 + point_index; + exp_histogram_point_values[point_index].count = 40 + point_index; + exp_histogram_point_values[point_index].scale = 4 + point_index; + + summary_points[point_index] = &summary_point_values[point_index]; + summary_point_values[point_index].time_unix_nano = 5000 + point_index; + summary_point_values[point_index].count = 50 + point_index; + summary_point_values[point_index].sum = 50.5 + point_index; + } + + gauge.n_data_points = 3; + gauge.data_points = gauge_points; + metrics[0].data_case = OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_GAUGE; + metrics[0].gauge = &gauge; + + sum.n_data_points = 3; + sum.data_points = sum_points; + sum.aggregation_temporality = + OPENTELEMETRY__PROTO__METRICS__V1__AGGREGATION_TEMPORALITY__AGGREGATION_TEMPORALITY_DELTA; + sum.is_monotonic = CMT_TRUE; + metrics[1].data_case = OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUM; + metrics[1].sum = ∑ + + histogram.n_data_points = 3; + histogram.data_points = histogram_points; + histogram.aggregation_temporality = + OPENTELEMETRY__PROTO__METRICS__V1__AGGREGATION_TEMPORALITY__AGGREGATION_TEMPORALITY_CUMULATIVE; + metrics[2].data_case = OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_HISTOGRAM; + metrics[2].histogram = &histogram; + + exp_histogram.n_data_points = 3; + exp_histogram.data_points = exp_histogram_points; + exp_histogram.aggregation_temporality = + OPENTELEMETRY__PROTO__METRICS__V1__AGGREGATION_TEMPORALITY__AGGREGATION_TEMPORALITY_DELTA; + metrics[3].data_case = + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_EXPONENTIAL_HISTOGRAM; + metrics[3].exponential_histogram = &exp_histogram; + + summary.n_data_points = 3; + summary.data_points = summary_points; + metrics[4].data_case = OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUMMARY; + metrics[4].summary = &summary; + + scope_metadata.name = "splitter scope"; + scope_metadata.version = "2.0.0"; + scope_metadata.dropped_attributes_count = 9; + scope.scope = &scope_metadata; + scope.schema_url = "https://example.com/scope/2.0.0"; + scope.n_metrics = 5; + scope.metrics = metric_entries; + scopes[0] = &scope; + + resource_metadata.dropped_attributes_count = 7; + resource.resource = &resource_metadata; + resource.schema_url = "https://example.com/resource/1.0.0"; + resource.n_scope_metrics = 1; + resource.scope_metrics = scopes; + resources[0] = &resource; + request.n_resource_metrics = 1; + request.resource_metrics = resources; + + payload_size = + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__get_packed_size( + &request); + payload = cfl_sds_create_size(payload_size); + TEST_CHECK(payload != NULL); + if (payload == NULL) { + return; + } + + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__pack( + &request, + (uint8_t *) payload); + cfl_sds_len_set(payload, payload_size); + + batches = cmt_encode_opentelemetry_split_payload(payload, + payload_size, + 2, + &result); + cfl_sds_destroy(payload); + TEST_CHECK(result == CMT_ENCODE_OPENTELEMETRY_SUCCESS); + TEST_CHECK(batches != NULL); + if (batches == NULL) { + return; + } + + TEST_CHECK(batches->count == 8); + total_data_points = 0; + + for (batch_index = 0; batch_index < batches->count; batch_index++) { + TEST_CHECK(batches->entries[batch_index].data_point_count == + (batch_index < 7 ? 2 : 1)); + + decoded = + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__unpack( + NULL, + cfl_sds_len(batches->entries[batch_index].payload), + (uint8_t *) batches->entries[batch_index].payload); + TEST_CHECK(decoded != NULL); + if (decoded == NULL) { + continue; + } + + TEST_CHECK(decoded->n_resource_metrics == 1); + decoded_resource = decoded->resource_metrics[0]; + TEST_CHECK(decoded_resource->resource != NULL); + TEST_CHECK(decoded_resource->schema_url != NULL); + if (decoded_resource->resource != NULL) { + TEST_CHECK(decoded_resource->resource->dropped_attributes_count == 7); + } + if (decoded_resource->schema_url != NULL) { + TEST_CHECK(strcmp(decoded_resource->schema_url, + "https://example.com/resource/1.0.0") == 0); + } + + TEST_CHECK(decoded_resource->n_scope_metrics == 1); + decoded_scope = decoded_resource->scope_metrics[0]; + TEST_CHECK(decoded_scope->scope != NULL); + TEST_CHECK(decoded_scope->schema_url != NULL); + if (decoded_scope->scope != NULL) { + TEST_CHECK(strcmp(decoded_scope->scope->name, "splitter scope") == 0); + TEST_CHECK(strcmp(decoded_scope->scope->version, "2.0.0") == 0); + TEST_CHECK(decoded_scope->scope->dropped_attributes_count == 9); + } + if (decoded_scope->schema_url != NULL) { + TEST_CHECK(strcmp(decoded_scope->schema_url, + "https://example.com/scope/2.0.0") == 0); + } + + for (metric_index = 0; + metric_index < decoded_scope->n_metrics; + metric_index++) { + metric = decoded_scope->metrics[metric_index]; + type_index = -1; + + if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_GAUGE) { + type_index = 0; + for (point_index = 0; + point_index < metric->gauge->n_data_points; + point_index++) { + timestamp = metric->gauge->data_points[point_index]->time_unix_nano; + TEST_CHECK(metric->gauge->data_points[point_index]->value_case == + OPENTELEMETRY__PROTO__METRICS__V1__NUMBER_DATA_POINT__VALUE_AS_INT); + data_point_index = timestamp - 1000; + TEST_CHECK(data_point_index < 3); + if (data_point_index < 3) { + point_seen[type_index][data_point_index]++; + } + } + total_data_points += metric->gauge->n_data_points; + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUM) { + type_index = 1; + TEST_CHECK(metric->sum->aggregation_temporality == sum.aggregation_temporality); + TEST_CHECK(metric->sum->is_monotonic == sum.is_monotonic); + for (point_index = 0; + point_index < metric->sum->n_data_points; + point_index++) { + timestamp = metric->sum->data_points[point_index]->time_unix_nano; + data_point_index = timestamp - 2000; + TEST_CHECK(data_point_index < 3); + if (data_point_index < 3) { + point_seen[type_index][data_point_index]++; + } + } + total_data_points += metric->sum->n_data_points; + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_HISTOGRAM) { + type_index = 2; + TEST_CHECK(metric->histogram->aggregation_temporality == + histogram.aggregation_temporality); + for (point_index = 0; + point_index < metric->histogram->n_data_points; + point_index++) { + timestamp = metric->histogram->data_points[point_index]->time_unix_nano; + data_point_index = timestamp - 3000; + TEST_CHECK(data_point_index < 3); + if (data_point_index < 3) { + point_seen[type_index][data_point_index]++; + } + } + total_data_points += metric->histogram->n_data_points; + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_EXPONENTIAL_HISTOGRAM) { + type_index = 3; + TEST_CHECK(metric->exponential_histogram->aggregation_temporality == + exp_histogram.aggregation_temporality); + for (point_index = 0; + point_index < metric->exponential_histogram->n_data_points; + point_index++) { + timestamp = metric->exponential_histogram->data_points[point_index]->time_unix_nano; + data_point_index = timestamp - 4000; + TEST_CHECK(data_point_index < 3); + if (data_point_index < 3) { + point_seen[type_index][data_point_index]++; + } + } + total_data_points += metric->exponential_histogram->n_data_points; + } + else if (metric->data_case == + OPENTELEMETRY__PROTO__METRICS__V1__METRIC__DATA_SUMMARY) { + type_index = 4; + for (point_index = 0; + point_index < metric->summary->n_data_points; + point_index++) { + timestamp = metric->summary->data_points[point_index]->time_unix_nano; + data_point_index = timestamp - 5000; + TEST_CHECK(data_point_index < 3); + if (data_point_index < 3) { + point_seen[type_index][data_point_index]++; + } + } + total_data_points += metric->summary->n_data_points; + } + + TEST_CHECK(type_index >= 0); + if (type_index >= 0) { + type_seen[type_index]++; + TEST_CHECK(strcmp(metric->name, metric_names[type_index]) == 0); + TEST_CHECK(strcmp(metric->description, "batch preservation test") == 0); + TEST_CHECK(strcmp(metric->unit, "1") == 0); + } + } + + opentelemetry__proto__collector__metrics__v1__export_metrics_service_request__free_unpacked( + decoded, + NULL); + } + + TEST_CHECK(total_data_points == 15); + for (type_index = 0; type_index < 5; type_index++) { + TEST_CHECK(type_seen[type_index] == 2); + for (point_index = 0; point_index < 3; point_index++) { + TEST_CHECK(point_seen[type_index][point_index] == 1); + } + } + + cmt_encode_opentelemetry_destroy_batches(batches); +} + TEST_LIST = { {"opentelemetry_api_full_roundtrip_with_msgpack", test_opentelemetry_api_full_roundtrip_with_msgpack}, {"opentelemetry_encode_multi_resource_scope_containers", test_opentelemetry_encode_multi_resource_scope_containers}, @@ -2009,5 +2592,8 @@ TEST_LIST = { {"opentelemetry_missing_metric_name_rejected", test_opentelemetry_missing_metric_name_rejected}, {"opentelemetry_missing_metric_data_rejected", test_opentelemetry_missing_metric_data_rejected}, {"opentelemetry_omitted_null_key_label_encoded", test_opentelemetry_omitted_null_key_label_encoded}, + {"opentelemetry_data_point_batches", test_opentelemetry_data_point_batches}, + {"opentelemetry_batches_empty_context", test_opentelemetry_batches_empty_context}, + {"opentelemetry_batches_all_metric_types", test_opentelemetry_batches_all_metric_types}, { 0 } };