From 81b565686e5817acd04c0fa6b9da35eb0e41f4bf Mon Sep 17 00:00:00 2001 From: Pranav Sathyanarayanan Date: Thu, 19 Mar 2026 14:31:38 -0400 Subject: [PATCH 1/2] Feature: Support jsontemplate in http-bulk --- CMakeLists.txt | 10 ++ README.md | 31 +++++ http-bulk-payload.cpp | 210 +++++++++++++++++++++++++++++++ http-bulk-payload.h | 25 ++++ http-bulk.cpp | 69 ++++++---- http-bulk.schema.json | 4 + tests/http-bulk-payload-test.cpp | 84 +++++++++++++ 7 files changed, 410 insertions(+), 23 deletions(-) create mode 100644 http-bulk-payload.cpp create mode 100644 http-bulk-payload.h create mode 100644 tests/http-bulk-payload-test.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 02e5379..37c52bd 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -32,6 +32,7 @@ SET(CMAKE_CXX_FLAGS_RELEASE "-std=c++17 -Wall -Wvla -Wshadow -Wconversion -Wno-s ADD_LIBRARY(http-bulk SHARED http-bulk.cpp + http-bulk-payload.cpp ) TARGET_LINK_LIBRARIES(http-bulk @@ -41,6 +42,15 @@ TARGET_LINK_LIBRARIES(http-bulk SET_TARGET_PROPERTIES(http-bulk PROPERTIES PREFIX "") +ENABLE_TESTING() + +ADD_EXECUTABLE(http-bulk-payload-test + http-bulk-payload.cpp + tests/http-bulk-payload-test.cpp +) + +ADD_TEST(NAME http-bulk-payload-test COMMAND http-bulk-payload-test) + INCLUDE_DIRECTORIES( /opt/halon/include ) diff --git a/README.md b/README.md index a59a61c..0cac8dd 100644 --- a/README.md +++ b/README.md @@ -117,8 +117,39 @@ plugins: headers: - "Content-Type: text/csv" - "Authorization: Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==" + - id: temporal + concurrency: 1 + path: /var/log/halon/temporal.jlog + format: jsonarray + jsontemplate: | + { + "workflowType": { "name": "handleHalonEmails" }, + "taskQueue": { "name": "lbx-live-halon-workflows", "kind": "TASK_QUEUE_KIND_NORMAL" }, + "identity": "halon-http-bulk", + "requestId": $UUIDV7, + "workflowTaskTimeout": "10s", + "input": { + "payloads": [ + { + "metadata": { "encoding": "anNvbi9wbGFpbg==" }, + "data": $DATA_BASE64 + } + ] + } + } + url: "http://1.2.3.4:7233/api/v1/namespaces/default/workflows" + timeout: 30 + max_items: 500 + tls_verify: true + headers: + - "Authorization: Bearer secret" ``` +When `jsontemplate` is set, the request body is rendered as JSON and the following raw template tags are supported: + +* `$UUIDV7` inserts a generated UUIDv7 JSON string. +* `$DATA_BASE64` inserts a base64-encoded JSON string of the current batch rendered as a canonical JSON array. + ### Operation of `min_items`, `max_items` and `interval` During busy periods, when there are many queued events, batches of size `max_items` will be sent. diff --git a/http-bulk-payload.cpp b/http-bulk-payload.cpp new file mode 100644 index 0000000..4707af7 --- /dev/null +++ b/http-bulk-payload.cpp @@ -0,0 +1,210 @@ +#include "http-bulk-payload.h" + +#include +#include +#include +#include +#include +#include + +namespace http_bulk +{ +namespace +{ +constexpr char UUIDV7_TAG[] = "$UUIDV7"; +constexpr char DATA_BASE64_TAG[] = "$DATA_BASE64"; + +std::string render_batch(payload_format format, + const std::vector& items, + bool wrap_json_array) +{ + std::string payload; + for (size_t i = 0; i < items.size(); ++i) + { + if (format == payload_format::jsonarray && wrap_json_array && i != 0) + payload += ","; + payload += items[i]; + if (format == payload_format::ndjson) + payload += "\n"; + } + + if (format == payload_format::jsonarray && wrap_json_array) + return "[" + payload + "]"; + return payload; +} + +std::string json_quote(const std::string& value) +{ + std::string out; + out.reserve(value.size() + 2); + out += "\""; + for (unsigned char ch : value) + { + switch (ch) + { + case '\\': + out += "\\\\"; + break; + case '\"': + out += "\\\""; + break; + case '\b': + out += "\\b"; + break; + case '\f': + out += "\\f"; + break; + case '\n': + out += "\\n"; + break; + case '\r': + out += "\\r"; + break; + case '\t': + out += "\\t"; + break; + default: + if (ch < 0x20) + { + static constexpr char hex[] = "0123456789abcdef"; + out += "\\u00"; + out += hex[(ch >> 4) & 0x0F]; + out += hex[ch & 0x0F]; + } + else + out += static_cast(ch); + break; + } + } + out += "\""; + return out; +} + +std::string replace_template_tags(const std::string& json_template, + const std::string& json_array_payload) +{ + std::string rendered; + rendered.reserve(json_template.size() + json_array_payload.size()); + + size_t pos = 0; + while (pos < json_template.size()) + { + size_t uuid_pos = json_template.find(UUIDV7_TAG, pos); + size_t data_pos = json_template.find(DATA_BASE64_TAG, pos); + size_t tag_pos = std::min(uuid_pos, data_pos); + + if (tag_pos == std::string::npos) + { + rendered.append(json_template, pos, std::string::npos); + break; + } + + rendered.append(json_template, pos, tag_pos - pos); + if (tag_pos == uuid_pos) + { + rendered += json_quote(generate_uuid_v7()); + pos = tag_pos + sizeof(UUIDV7_TAG) - 1; + } + else + { + rendered += json_quote(base64_encode(json_array_payload)); + pos = tag_pos + sizeof(DATA_BASE64_TAG) - 1; + } + } + + return rendered; +} +} + +std::string render_json_array(const std::vector& items) +{ + return render_batch(payload_format::jsonarray, items, true); +} + +std::string generate_uuid_v7() +{ + std::array bytes{}; + + const auto now = std::chrono::time_point_cast( + std::chrono::system_clock::now()); + const uint64_t timestamp = + static_cast(now.time_since_epoch().count()); + + // 48-bit big-endian Unix timestamp in milliseconds + for (size_t i = 0; i < 6; ++i) + bytes[i] = static_cast((timestamp >> ((5 - i) * 8)) & 0xFF); + + static thread_local std::mt19937_64 rng(std::random_device{}()); + static thread_local std::uniform_int_distribution dist(0, 255); + + for (size_t i = 6; i < bytes.size(); ++i) + bytes[i] = static_cast(dist(rng)); + + // Set version (7) + bytes[6] = static_cast((bytes[6] & 0x0F) | 0x70); + + // Set variant (10xxxxxx) + bytes[8] = static_cast((bytes[8] & 0x3F) | 0x80); + + static constexpr char hex[] = "0123456789abcdef"; + std::string uuid; + uuid.reserve(36); + + for (size_t i = 0; i < bytes.size(); ++i) + { + if (i == 4 || i == 6 || i == 8 || i == 10) + uuid.push_back('-'); + uuid.push_back(hex[(bytes[i] >> 4) & 0x0F]); + uuid.push_back(hex[bytes[i] & 0x0F]); + } + + return uuid; +} + +std::string base64_encode(const std::string& input) +{ + static constexpr char alphabet[] = + "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + + std::string out; + out.reserve(((input.size() + 2) / 3) * 4); + + uint32_t value = 0; + int bits = -6; + for (unsigned char ch : input) + { + value = (value << 8) | ch; + bits += 8; + while (bits >= 0) + { + out += alphabet[(value >> bits) & 0x3Fu]; + bits -= 6; + } + } + + if (bits > -6) + out += alphabet[((value << 8) >> (bits + 8)) & 0x3Fu]; + while (out.size() % 4 != 0) + out += '='; + return out; +} + +std::string render_payload(payload_format format, + const std::vector& items, + bool wrap_json_array, + const std::string& preamble, + const std::string& postamble, + const std::string& json_template) +{ + std::string payload = preamble; + if (!json_template.empty()) + { + // Template substitutions are based on a canonical JSON array payload. + payload += replace_template_tags(json_template, render_json_array(items)); + } + else + payload += render_batch(format, items, wrap_json_array); + payload += postamble; + return payload; +} +} diff --git a/http-bulk-payload.h b/http-bulk-payload.h new file mode 100644 index 0000000..8a41a0d --- /dev/null +++ b/http-bulk-payload.h @@ -0,0 +1,25 @@ +#pragma once + +#include +#include + +namespace http_bulk +{ +enum class payload_format +{ + ndjson, + jsonarray, + custom +}; + +std::string render_payload(payload_format format, + const std::vector& items, + bool wrap_json_array, + const std::string& preamble, + const std::string& postamble, + const std::string& json_template); + +std::string render_json_array(const std::vector& items); +std::string generate_uuid_v7(); +std::string base64_encode(const std::string& input); +} diff --git a/http-bulk.cpp b/http-bulk.cpp index 5229049..534d167 100644 --- a/http-bulk.cpp +++ b/http-bulk.cpp @@ -13,6 +13,9 @@ #include #include #include +#include + +#include "http-bulk-payload.h" // Halon > 6.1 is linked against libcurl with this feature #ifndef CURLOPT_AWS_SIGV4 @@ -61,6 +64,7 @@ struct bulkQueue std::string url; std::string preamble, postamble; + std::string jsonTemplate; format_t format = F_JSONARRAY; bool tls_verify = true; std::vector headers; @@ -97,6 +101,20 @@ struct curlResult { std::string result; }; +static http_bulk::payload_format to_payload_format(format_t format) +{ + switch (format) + { + case F_NDJSON: + return http_bulk::payload_format::ndjson; + case F_JSONARRAY: + return http_bulk::payload_format::jsonarray; + case F_CUSTOM: + return http_bulk::payload_format::custom; + } + return http_bulk::payload_format::custom; +} + static void curl_multi() { pthread_setname_np(pthread_self(), "p/http-bulk/req"); @@ -212,17 +230,24 @@ static void subscriber(std::shared_ptr queue) auto lastSend = std::chrono::steady_clock::now(); struct curl_slist* hdrs = nullptr; - switch (queue->format) + if (!queue->jsonTemplate.empty()) { - case F_NDJSON: - hdrs = curl_slist_append(hdrs, "Content-Type: application/x-ndjson"); - break; - case F_JSONARRAY: - hdrs = curl_slist_append(hdrs, "Content-Type: application/json"); - break; - case F_CUSTOM: - /* no header */ - break; + hdrs = curl_slist_append(hdrs, "Content-Type: application/json"); + } + else + { + switch (queue->format) + { + case F_NDJSON: + hdrs = curl_slist_append(hdrs, "Content-Type: application/x-ndjson"); + break; + case F_JSONARRAY: + hdrs = curl_slist_append(hdrs, "Content-Type: application/json"); + break; + case F_CUSTOM: + /* no header */ + break; + } } for (const auto & i : queue->headers) hdrs = curl_slist_append(hdrs, i.c_str()); @@ -275,11 +300,7 @@ static void subscriber(std::shared_ptr queue) send: lastSend = std::chrono::steady_clock::now(); - - std::string payload = queue->preamble; - - if (queue->maxItems > 1 && queue->format == F_JSONARRAY) - payload += "["; + std::vector payloadItems; size_t items = 0; for (; items < std::min(queue->maxItems, (size_t)count); items++, JLOG_ID_ADVANCE(&begin)) @@ -291,17 +312,17 @@ static void subscriber(std::shared_ptr queue) syslog(LOG_CRIT, "http_bulk: jlog_ctx_read_message failed: %d %s", jlog_ctx_err(queue->readerContext), jlog_ctx_err_string(queue->readerContext)); return; } - if (queue->maxItems > 1 && items != 0 && queue->format == F_JSONARRAY) - payload += ","; - payload += std::string((char*)m.mess, m.mess_len); - if (queue->format == F_NDJSON) - payload += "\n"; + payloadItems.emplace_back((char*)m.mess, m.mess_len); if (queue->oneItem) break; } - if (queue->maxItems > 1 && queue->format == F_JSONARRAY) - payload += "]"; - payload += queue->postamble; + std::string payload = http_bulk::render_payload( + to_payload_format(queue->format), + payloadItems, + queue->maxItems > 1, + queue->preamble, + queue->postamble, + queue->jsonTemplate); auto h = new curlResult; @@ -643,6 +664,7 @@ bool Halon_init(HalonInitContext* hic) const char* tls_verify = HalonMTA_config_string_get(HalonMTA_config_object_get(queue, "tls_verify"), nullptr); const char* preamble = HalonMTA_config_string_get(HalonMTA_config_object_get(queue, "preamble"), nullptr); const char* postamble = HalonMTA_config_string_get(HalonMTA_config_object_get(queue, "postamble"), nullptr); + const char* jsontemplate = HalonMTA_config_string_get(HalonMTA_config_object_get(queue, "jsontemplate"), nullptr); const char* username = HalonMTA_config_string_get(HalonMTA_config_object_get(queue, "username"), nullptr); const char* password = HalonMTA_config_string_get(HalonMTA_config_object_get(queue, "password"), nullptr); const char* aws_sigv4 = HalonMTA_config_string_get(HalonMTA_config_object_get(queue, "aws_sigv4"), nullptr); @@ -707,6 +729,7 @@ bool Halon_init(HalonInitContext* hic) x->maxInterval = !maxinterval ? 0 : strtoul(maxinterval, nullptr, 10); x->preamble = preamble ? preamble : ""; x->postamble = postamble ? postamble : ""; + x->jsonTemplate = jsontemplate ? jsontemplate : ""; x->username = username ? username : ""; x->password = password ? password : ""; x->aws_sigv4 = aws_sigv4 ? aws_sigv4 : ""; diff --git a/http-bulk.schema.json b/http-bulk.schema.json index e1352df..bf91f28 100644 --- a/http-bulk.schema.json +++ b/http-bulk.schema.json @@ -35,6 +35,10 @@ "default": "jsonarray", "description": "JSON bulk format" }, + "jsontemplate": { + "type": "string", + "description": "JSON request template. Supports raw tags $UUIDV7 and $DATA_BASE64" + }, "min_items": { "type": "number", "description": "Minimum items to send in bulk request", diff --git a/tests/http-bulk-payload-test.cpp b/tests/http-bulk-payload-test.cpp new file mode 100644 index 0000000..7375783 --- /dev/null +++ b/tests/http-bulk-payload-test.cpp @@ -0,0 +1,84 @@ +#include "../http-bulk-payload.h" + +#include +#include +#include +#include +#include + +namespace +{ +void expect(bool condition, const std::string& message) +{ + if (!condition) + { + std::cerr << message << "\n"; + std::abort(); + } +} + +void test_jsonarray_legacy_single_item() +{ + const std::vector items { R"({"id":1})" }; + const std::string payload = http_bulk::render_payload( + http_bulk::payload_format::jsonarray, + items, + false, + "", + "", + ""); + expect(payload == R"({"id":1})", + "jsonarray without wrapping should preserve legacy single-item output"); +} + +void test_json_template_base64_uses_jsonarray() +{ + const std::vector items { R"({"id":1})", R"({"id":2})" }; + const std::string payload = http_bulk::render_payload( + http_bulk::payload_format::ndjson, + items, + false, + "", + "", + R"({"input":{"payloads":[{"data":$DATA_BASE64}]}})"); + const std::string expected = R"({"input":{"payloads":[{"data":")" + + http_bulk::base64_encode(R"([{"id":1},{"id":2}])") + + R"("}]}})"; + expect(payload == expected, + "json template should inject base64 of the canonical json array payload"); +} + +void test_json_template_adds_uuidv7_json_string() +{ + const std::vector items { R"({"id":1})" }; + const std::string payload = http_bulk::render_payload( + http_bulk::payload_format::jsonarray, + items, + false, + "", + "", + R"({"requestId":$UUIDV7})"); + + expect(payload.size() == 52, "uuidv7 payload length should match the JSON string shape"); + expect(payload.rfind(R"({"requestId":")", 0) == 0, + "uuidv7 template should inject a JSON string"); + expect(payload.back() == '}', "uuidv7 payload should be valid JSON text"); + + const std::string uuid = payload.substr(14, 36); + expect(uuid[8] == '-' && uuid[13] == '-' && uuid[18] == '-' && uuid[23] == '-', + "uuidv7 should preserve canonical dashes"); + expect(uuid[14] == '7', "uuidv7 version nibble should be 7"); + + const char variant = static_cast(std::tolower(static_cast(uuid[19]))); + expect(variant >= '8' && variant <= 'b', + "uuidv7 variant nibble should be RFC 4122 compatible"); +} +} + +int main() +{ + test_jsonarray_legacy_single_item(); + test_json_template_base64_uses_jsonarray(); + test_json_template_adds_uuidv7_json_string(); + return 0; +} From a8f91bae7de3350810d6684ba56f23b19ff392d5 Mon Sep 17 00:00:00 2001 From: Pranav Sathyanarayanan Date: Thu, 19 Mar 2026 16:43:31 -0400 Subject: [PATCH 2/2] Update README.md Remove "lbx-live" annotations --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 0cac8dd..dcdfff7 100644 --- a/README.md +++ b/README.md @@ -124,7 +124,7 @@ plugins: jsontemplate: | { "workflowType": { "name": "handleHalonEmails" }, - "taskQueue": { "name": "lbx-live-halon-workflows", "kind": "TASK_QUEUE_KIND_NORMAL" }, + "taskQueue": { "name": "halon-workflows", "kind": "TASK_QUEUE_KIND_NORMAL" }, "identity": "halon-http-bulk", "requestId": $UUIDV7, "workflowTaskTimeout": "10s", @@ -137,7 +137,7 @@ plugins: ] } } - url: "http://1.2.3.4:7233/api/v1/namespaces/default/workflows" + url: "http://1.2.3.4:7233/api/v1/namespaces/default/workflows/halon-http-bulk" timeout: 30 max_items: 500 tls_verify: true