diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 0b387f1..27d414c 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -103,6 +103,31 @@ Say "setup omc" or run `/oh-my-claudecode:omc-setup`. +## Mandatory Codebase Discovery Preflight + +For any task that requires finding files, symbols, classes, modules, call chains, +cross-file relationships, or unknown implementation locations, the first discovery +step MUST be Codebase Memory. + +Required sequence: + +1. If Codebase Memory tools are not currently visible, resolve them first: + `ToolSearch("select:mcp__codebase-memory__index_status")`. +2. Call `mcp__codebase-memory__index_status`. +3. If the project is indexed, use: + - `mcp__codebase-memory__search_graph` for symbols/classes/modules/files; + - `mcp__codebase-memory__trace_path` for call chains/dependencies; + - `mcp__codebase-memory__get_architecture` for module structure; + - `mcp__codebase-memory__get_code_snippet` for targeted code. +4. If the project is not indexed or path is ambiguous, use + `mcp__codebase-memory__list_projects` and/or `mcp__codebase-memory__index_repository`. +5. Only if Codebase Memory is unavailable or fails after retry, fall back to + Glob/Grep/Read/LSP. + +Do not start first-pass codebase discovery with `git status`, Bash, Glob, Grep, +or Read. `git status` is allowed for worktree safety before edits/staging, but it +does not satisfy discovery preflight and must not replace Codebase Memory. + ## Specific Overrides - "Delegate" → always route via `~/.claude/rules/delegation.md` routing table; never decide ad-hoc. diff --git a/.claude/agents/architect.md b/.claude/agents/architect.md index 1bcbb88..3b04dbd 100644 --- a/.claude/agents/architect.md +++ b/.claude/agents/architect.md @@ -24,7 +24,10 @@ You are Architect. Analyze code, diagnose bugs, and provide actionable architect ## Investigation Protocol -1) Gather context first (MANDATORY): Glob project structure, Grep/Read relevant implementations, check manifests, find tests. Execute in parallel. +1) Gather context first (MANDATORY): + for non-trivial codebase questions, run Codebase Memory preflight before Glob/Grep/Read: + `index_status` → `search_graph` / `trace_path` / `get_architecture` → targeted Read/LSP. + Use Glob/Grep only as fallback or precision confirmation. 2) For debugging: Read error messages completely. Check recent changes (git log/blame). Find working examples. Compare broken vs working. 3) Form hypothesis and document BEFORE looking deeper. 4) Cross-reference hypothesis against actual code. Cite file:line for every claim. @@ -35,6 +38,12 @@ You are Architect. Analyze code, diagnose bugs, and provide actionable architect ## Tool Usage +- **Codebase Memory — primary for codebase discovery**: + `mcp__codebase-memory__index_status`, + `mcp__codebase-memory__search_graph`, + `mcp__codebase-memory__trace_path`, + `mcp__codebase-memory__get_architecture`, + `mcp__codebase-memory__get_code_snippet` - **Core**: Glob, Grep, Read, Bash (git blame/log) - **Context-mode**: ctx_search, ctx_execute, ctx_execute_file, ctx_batch_execute, ctx_fetch_and_index - **LSP**: lsp_diagnostics, lsp_diagnostics_directory, lsp_hover, lsp_goto_definition, lsp_find_references, lsp_document_symbols, lsp_workspace_symbols, lsp_code_actions, lsp_rename, lsp_servers diff --git a/.claude/agents/debugger.md b/.claude/agents/debugger.md index 7f52179..e3762d9 100644 --- a/.claude/agents/debugger.md +++ b/.claude/agents/debugger.md @@ -37,6 +37,7 @@ You are Debugger. Trace bugs to their root cause and recommend minimal fixes. Ge ## Tool Usage +- **Codebase Memory**: use `index_status` → `search_graph` / `trace_path` / `get_code_snippet` before Grep/Read when tracing call chains or cross-file dependencies in unfamiliar code. - **Core**: Read, Grep, Bash (git blame/log, build commands), Edit (minimal fixes only) - **Context-mode**: ctx_search, ctx_execute, ctx_execute_file, ctx_batch_execute, ctx_fetch_and_index - **LSP**: lsp_diagnostics, lsp_diagnostics_directory (preferred over CLI for TypeScript), lsp_hover, lsp_goto_definition, lsp_find_references, lsp_document_symbols, lsp_workspace_symbols diff --git a/.claude/agents/explore.md b/.claude/agents/explore.md index e2baee2..a4a348a 100644 --- a/.claude/agents/explore.md +++ b/.claude/agents/explore.md @@ -8,13 +8,28 @@ disallowedTools: Write, Edit You are Explorer. Find files, code patterns, and relationships in the codebase and return actionable results. Answer "where is X?", "which files contain Y?", "how does Z connect to W?" Not responsible for modifying code, implementing features, or external documentation search. Route external docs/literature requests to document-specialist. +## Codebase Discovery Protocol + +For non-trivial repository investigation, architecture questions, cross-file relationships, +call chains, refactors, or unknown implementation locations: + +1. First check `mcp__codebase-memory__index_status`. +2. If the project is not indexed, stale, or path is ambiguous, use + `mcp__codebase-memory__list_projects` and/or `mcp__codebase-memory__index_repository`. +3. Use `mcp__codebase-memory__search_graph` for symbols, classes, modules, and files. +4. Use `mcp__codebase-memory__trace_path` for call chains and dependency flow. +5. Use `mcp__codebase-memory__get_code_snippet` for targeted code snippets. +6. Use Glob/Grep/Read only after Codebase Memory, or as fallback if Codebase Memory is unavailable. + +Do not start non-trivial codebase discovery with Glob or Grep. + ## Constraints - Read-only: cannot create, modify, or delete files - Always use absolute paths (starting with /) - Return results as message text, never store in files - For symbol usage lookups requiring lsp_find_references, escalate to explore-high -- Launch 3+ parallel searches on first action, broad-to-narrow strategy +- For non-trivial codebase discovery, first run Codebase Memory preflight; then launch parallel searches if additional confirmation is needed. - Cross-validate across multiple tools (Grep vs Glob vs ast_grep_search) - Cap exploratory depth: stop after 2 rounds of diminishing returns - Medium effort: 3-5 parallel searches; thorough: 5-10; quick lookups: 1-2 @@ -28,6 +43,15 @@ You are Explorer. Find files, code patterns, and relationships in the codebase a ## Tools +- **Codebase Memory — primary for codebase discovery**: + `mcp__codebase-memory__index_status`, + `mcp__codebase-memory__list_projects`, + `mcp__codebase-memory__index_repository`, + `mcp__codebase-memory__search_graph`, + `mcp__codebase-memory__trace_path`, + `mcp__codebase-memory__get_code_snippet`, + `mcp__codebase-memory__get_architecture`, + `mcp__codebase-memory__search_code` - **Core**: Glob (file structure), Grep (text patterns), Read (targeted with offset/limit) - **Context-mode**: ctx_search, ctx_batch_execute, ctx_execute, ctx_execute_file, ctx_fetch_and_index - **LSP**: lsp_document_symbols, lsp_workspace_symbols, lsp_hover, lsp_goto_definition, lsp_find_references, lsp_diagnostics diff --git a/.claude/agents/tracer.md b/.claude/agents/tracer.md index d72f2f4..ccbbc72 100644 --- a/.claude/agents/tracer.md +++ b/.claude/agents/tracer.md @@ -57,6 +57,7 @@ You are not responsible for implementation, generic code review, generic summari - [Confirmation Bias](software-laws.md#confirmation-bias): actively seek disconfirming evidence for the leading hypothesis. Collect evidence against, not just for. ## Tools +**Codebase Memory**: use `index_status` → `search_graph` / `trace_path` / `get_code_snippet` before Grep/Read when tracing call chains or cross-file dependencies in unfamiliar code. **Core**: Read, Grep, Glob, Bash (focused evidence gathering) **Context-mode**: ctx_search, ctx_execute, ctx_batch_execute, ctx_execute_file, ctx_fetch_and_index **LSP**: lsp_diagnostics, lsp_diagnostics_directory, lsp_hover, lsp_goto_definition, lsp_find_references, lsp_document_symbols diff --git a/.claude/rules/delegation.md b/.claude/rules/delegation.md index b2b4bdc..74fe990 100644 --- a/.claude/rules/delegation.md +++ b/.claude/rules/delegation.md @@ -75,7 +75,7 @@ TaskCreate = conversation tracking only, NOT delegation. | Category | Agents | Primary | Fallback | |----------|--------|---------|----------| -| Analysis | explore, analyst, tracer, scientist | ctx_*, python_repl, Grep/Glob, session_search | Bash, DDG | +| Analysis / Architecture | explore, analyst, tracer, architect, scientist | Codebase Memory (`mcp__codebase-memory__*`), ctx_*, python_repl, Grep/Glob, session_search | Bash, DDG | | Implementation | executor, verifier, debugger, test-engineer | Edit/Write, LSP, ast_grep, ctx_execute, Bash (tests) | Bash, Grep, python_repl | | Review & Security | code-reviewer, security-reviewer | LSP, ast_grep_search, Grep, ctx_execute_file | Read, Bash | | Specialist | document-specialist, architect, writer, git-master | context7, GitHub, LSP, ctx_execute_file, Read | DDG, Fetch, gh CLI | diff --git a/.claude/rules/tool-priority.md b/.claude/rules/tool-priority.md index a971803..35b66bc 100644 --- a/.claude/rules/tool-priority.md +++ b/.claude/rules/tool-priority.md @@ -106,6 +106,19 @@ LSP, OMC State/Notepad, and AST tools listed as `lsp_*`, `state_*`, `notepad_*`, ## Codebase Memory Usage Policy +This policy applies to the orchestrator and to every subagent that performs codebase discovery. +Subagents must not assume the orchestrator has already completed discovery unless the prompt +explicitly provides the relevant Codebase Memory results, qualified names, or file paths. + +If the orchestrator delegates discovery to a subagent, the delegation prompt must either: +1. include explicit Codebase Memory preflight instructions, or +2. provide already verified Codebase Memory results and tell the subagent to continue from them. + +The first observable discovery tool call must be Codebase Memory, not Bash, Glob, +Grep, Read, or LSP. Planning text is not enough: if the reasoning says Codebase +Memory will be used, the next discovery action must actually call Codebase Memory +or explicitly report that the tool is unavailable. + Use Codebase Memory before Grep/Read/LSP when the task asks to: - find where behavior is implemented; - understand module architecture; diff --git a/AGENTS.md b/AGENTS.md index 29ccaee..b40befa 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -11,6 +11,17 @@ itself rather than on generic product or application architecture. - Use Conventional Commits: `type(scope): summary`. - Commit headers must be in English and include a descriptive body. +## Codebase Discovery + +For non-trivial codebase investigation, architecture questions, cross-file edits, +refactors, call chains, or unknown implementation locations, use Codebase Memory +before Grep/Glob/Read/LSP. + +Preferred sequence: +`index_status` → `search_graph` / `trace_path` / `get_code_snippet` → targeted Read/LSP. + +Grep/Glob are fallback or precision-confirmation tools, not first-pass architecture discovery. + ## Public Entry Headers Use the project umbrella headers instead of recreating include order manually: diff --git a/include/logit_cpp/logit/loggers/OtlpHttpLogger.hpp b/include/logit_cpp/logit/loggers/OtlpHttpLogger.hpp index 933f03e..f2b15df 100644 --- a/include/logit_cpp/logit/loggers/OtlpHttpLogger.hpp +++ b/include/logit_cpp/logit/loggers/OtlpHttpLogger.hpp @@ -12,6 +12,7 @@ #include "ILogger.hpp" #include "otlp/OtlpJsonFormatConfig.hpp" #include "otlp/OtlpJsonSerializer.hpp" +#include "otlp/OtlpPayloadSplitter.hpp" #ifndef KURLYK_WEBSOCKET_SUPPORT # define KURLYK_WEBSOCKET_SUPPORT 0 @@ -63,6 +64,7 @@ namespace logit { int request_timeout_sec = 3; long retry_attempts = 2; long retry_delay_ms = 250; + std::size_t max_payload_bytes = 1024 * 1024; bool drop_on_overflow = true; bool async = true; bool cancel_on_shutdown = false; @@ -116,10 +118,6 @@ namespace logit { if (!m_config.async) { std::vector batch; batch.push_back(item); - { - std::lock_guard lock(m_state->mutex); - ++m_state->http_in_flight; - } submit_batch_async(batch); try { m_client.wait_requests(); @@ -302,7 +300,6 @@ namespace logit { } m_state->space_cv.notify_all(); - ++m_state->http_in_flight; } if (!batch.empty()) { @@ -311,46 +308,70 @@ namespace logit { } } - /// \brief Submits one batch asynchronously to the configured OTLP endpoint. + /// \brief Submits one batch asynchronously, splitting into payload chunks. /// \param batch Batch to export. void submit_batch_async(const std::vector& batch) { - const std::string payload = build_otlp_logs_json_payload(batch, m_config.format); + auto chunks = build_otlp_logs_json_payload_chunks( + batch, m_config.format, m_config.max_payload_bytes); + + if (chunks.empty()) { + return; + } + kurlyk::Headers headers; headers.emplace("Content-Type", "application/json"); auto weak_state = std::weak_ptr(m_state); - bool submitted = false; - - try { - submitted = m_client.post( - m_config.path, {}, headers, payload, - [weak_state](kurlyk::HttpResponsePtr response) { - auto state = weak_state.lock(); - if (!state) { - return; - } - - std::lock_guard lock(state->mutex); - if (!response || response->status_code < 200 || response->status_code >= 300) { - state->failed_exports.fetch_add(1); - } - if (state->http_in_flight > 0) { - --state->http_in_flight; - } - state->cv.notify_all(); + + for (auto& chunk : chunks) { + { + std::unique_lock lock(m_state->mutex); + m_state->cv.wait(lock, [this]() { + return (m_state->stopping && m_config.cancel_on_shutdown) || + m_state->http_in_flight < m_config.max_in_flight_requests; }); - } catch (...) { - submitted = false; - } - if (!submitted) { - auto state = m_state; - std::lock_guard lock(state->mutex); - state->failed_exports.fetch_add(1); - if (state->http_in_flight > 0) { - --state->http_in_flight; + if (m_state->stopping && m_config.cancel_on_shutdown) { + m_state->failed_exports.fetch_add(1); + continue; + } + + ++m_state->http_in_flight; + } + + bool submitted = false; + + try { + submitted = m_client.post( + m_config.path, {}, headers, std::move(chunk), + [weak_state](kurlyk::HttpResponsePtr response) { + auto state = weak_state.lock(); + if (!state) { + return; + } + + std::lock_guard lock(state->mutex); + if (!response || response->status_code < 200 || response->status_code >= 300) { + state->failed_exports.fetch_add(1); + } + if (state->http_in_flight > 0) { + --state->http_in_flight; + } + state->cv.notify_all(); + }); + } catch (...) { + submitted = false; + } + + if (!submitted) { + auto state = m_state; + std::lock_guard lock(state->mutex); + state->failed_exports.fetch_add(1); + if (state->http_in_flight > 0) { + --state->http_in_flight; + } + state->cv.notify_all(); } - state->cv.notify_all(); } } diff --git a/include/logit_cpp/logit/loggers/OtlpPayloadLogger.hpp b/include/logit_cpp/logit/loggers/OtlpPayloadLogger.hpp index b9f4063..3b4c829 100644 --- a/include/logit_cpp/logit/loggers/OtlpPayloadLogger.hpp +++ b/include/logit_cpp/logit/loggers/OtlpPayloadLogger.hpp @@ -12,12 +12,14 @@ #include "ILogger.hpp" #include "otlp/OtlpJsonFormatConfig.hpp" #include "otlp/OtlpJsonSerializer.hpp" +#include "otlp/OtlpPayloadSplitter.hpp" #include #include #include #include #include +#include #include #include #include @@ -38,6 +40,7 @@ namespace logit { OtlpJsonFormatConfig format; std::function on_payload; bool async = true; + std::size_t max_payload_bytes = 1024 * 1024; std::size_t max_batch_size = 256; std::size_t max_queue_size = 1024; bool drop_on_overflow = true; @@ -88,13 +91,16 @@ namespace logit { } std::vector batch; batch.push_back(item); - std::string payload = build_otlp_logs_json_payload(batch, m_config.format); - try { - if (m_config.on_payload) { - m_config.on_payload(std::move(payload)); + auto chunks = build_otlp_logs_json_payload_chunks( + batch, m_config.format, m_config.max_payload_bytes); + if (m_config.on_payload) { + for (auto& chunk : chunks) { + try { + m_config.on_payload(std::move(chunk)); + } catch (...) { + ++m_failed_exports; + } } - } catch (...) { - ++m_failed_exports; } return; } @@ -269,13 +275,16 @@ namespace logit { } if (!batch.empty()) { - std::string payload = build_otlp_logs_json_payload(batch, m_config.format); - try { - if (m_config.on_payload) { - m_config.on_payload(std::move(payload)); + auto chunks = build_otlp_logs_json_payload_chunks( + batch, m_config.format, m_config.max_payload_bytes); + if (m_config.on_payload) { + for (auto& chunk : chunks) { + try { + m_config.on_payload(std::move(chunk)); + } catch (...) { + ++m_failed_exports; + } } - } catch (...) { - ++m_failed_exports; } } diff --git a/include/logit_cpp/logit/loggers/otlp/OtlpPayloadSplitter.hpp b/include/logit_cpp/logit/loggers/otlp/OtlpPayloadSplitter.hpp new file mode 100644 index 0000000..90de01f --- /dev/null +++ b/include/logit_cpp/logit/loggers/otlp/OtlpPayloadSplitter.hpp @@ -0,0 +1,65 @@ +#pragma once +#ifndef _LOGIT_OTLP_PAYLOAD_SPLITTER_HPP_INCLUDED +#define _LOGIT_OTLP_PAYLOAD_SPLITTER_HPP_INCLUDED + +/// \file OtlpPayloadSplitter.hpp +/// \brief Splits oversized OTLP log batches into multiple JSON payload chunks. + +#include "OtlpJsonSerializer.hpp" +#include +#include + +namespace logit { + + /// \brief Builds OTLP JSON payload chunks from a batch, splitting when a chunk + /// would exceed max_payload_bytes. + /// \param batch Log items to serialize. + /// \param format Serialization configuration. + /// \param max_payload_bytes Maximum serialized size per chunk (0 = no splitting). + /// \return Vector of JSON payload strings, one per chunk. + inline std::vector build_otlp_logs_json_payload_chunks( + const std::vector& batch, + const OtlpJsonFormatConfig& format, + std::size_t max_payload_bytes) { + if (max_payload_bytes == 0) { + std::vector chunks; + chunks.push_back(build_otlp_logs_json_payload(batch, format)); + return chunks; + } + + if (batch.empty()) { + return {}; + } + + std::vector chunks; + std::vector current_payload; + + for (std::size_t i = 0; i < batch.size(); ++i) { + std::vector candidate_payload = current_payload; + candidate_payload.push_back(batch[i]); + + std::string candidate_json = build_otlp_logs_json_payload(candidate_payload, format); + + if (candidate_json.size() <= max_payload_bytes) { + current_payload = std::move(candidate_payload); + } else if (current_payload.empty()) { + // Single oversized record: emit it as its own chunk anyway. + chunks.push_back(std::move(candidate_json)); + } else { + // Close current chunk and start a new one with this record. + chunks.push_back(build_otlp_logs_json_payload(current_payload, format)); + current_payload.clear(); + current_payload.push_back(batch[i]); + } + } + + if (!current_payload.empty()) { + chunks.push_back(build_otlp_logs_json_payload(current_payload, format)); + } + + return chunks; + } + +} // namespace logit + +#endif // _LOGIT_OTLP_PAYLOAD_SPLITTER_HPP_INCLUDED diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 0ca98c1..711aa27 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -52,6 +52,7 @@ else() otlp_http_logger_integration_test.cpp otlp_http_logger_callback_test.cpp otlp_json_serializer_test.cpp + otlp_payload_splitter_test.cpp otlp_structured_attributes_test.cpp otlp_payload_logger_test.cpp prometheus_text_serializer_test.cpp @@ -86,6 +87,7 @@ else() list(REMOVE_ITEM TEST_SOURCES otlp_http_logger_integration_test.cpp) list(REMOVE_ITEM TEST_SOURCES otlp_http_logger_callback_test.cpp) list(REMOVE_ITEM TEST_SOURCES otlp_structured_attributes_test.cpp) + list(REMOVE_ITEM TEST_SOURCES otlp_payload_splitter_test.cpp) list(REMOVE_ITEM TEST_SOURCES otlp_payload_logger_test.cpp) endif() if(NOT LOGIT_WITH_PROMETHEUS) diff --git a/tests/otlp_http_logger_callback_test.cpp b/tests/otlp_http_logger_callback_test.cpp index 3d86dc7..0a12fc5 100644 --- a/tests/otlp_http_logger_callback_test.cpp +++ b/tests/otlp_http_logger_callback_test.cpp @@ -20,6 +20,7 @@ struct RequestCounter { std::mutex mutex; std::condition_variable cv; std::atomic count{0}; + std::vector bodies; std::string last_body; bool delay_response = false; int delay_ms = 0; @@ -60,6 +61,7 @@ void start_server(HttpServer& server, std::thread& thread, RequestCounter& count { std::lock_guard lock(counter.mutex); counter.last_body = request->content.string(); + counter.bodies.push_back(counter.last_body); counter.count.fetch_add(1); } counter.cv.notify_all(); @@ -357,6 +359,58 @@ int main() { stop_server(server, server_thread); } + // Test h: payload splitting produces multiple POSTs with small max_payload_bytes + { + RequestCounter counter; + HttpServer server; + std::thread server_thread; + start_server(server, server_thread, counter, port); + + logit::OtlpHttpLogger::Config config; + config.host = "http://127.0.0.1:" + std::to_string(port); + config.path = "/v1/logs"; + config.format.service_name = "http-split-test"; + config.max_batch_size = 256; + config.max_payload_bytes = 1024; + config.export_interval_ms = 50; + config.request_timeout_sec = 5; + config.max_in_flight_requests = 8; + + auto logger = std::unique_ptr(new logit::OtlpHttpLogger(config)); + + for (int i = 0; i < 50; ++i) { + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL + i, + "test.cpp", 200 + i, "test_func", "http split test", "", + -1, false, false, false); + logger->log(record, "http split payload test message number " + std::to_string(i)); + } + + logger->wait(); + logger->shutdown(); + + { + std::unique_lock lock(counter.mutex); + counter.cv.wait_for(lock, std::chrono::seconds(3), [&counter]() { + return counter.count.load() >= 2; + }); + } + + assert(counter.count.load() > 1); + + int body_count = 0; + for (const auto& body : counter.bodies) { + std::size_t pos = 0; + while ((pos = body.find("\"body\"", pos)) != std::string::npos) { + ++body_count; + ++pos; + } + } + assert(body_count == 50); + + stop_server(server, server_thread); + } + return 0; } diff --git a/tests/otlp_payload_logger_test.cpp b/tests/otlp_payload_logger_test.cpp index f2effd2..0f0c4af 100644 --- a/tests/otlp_payload_logger_test.cpp +++ b/tests/otlp_payload_logger_test.cpp @@ -1,4 +1,5 @@ -#include +#include +#include #ifdef LOGIT_WITH_OTLP @@ -13,7 +14,7 @@ namespace { -struct PayloadCollector { +struct Collector { std::mutex mutex; std::condition_variable cv; std::atomic count{0}; @@ -25,32 +26,30 @@ struct PayloadCollector { int main() { // Test a: sync mode callback receives valid JSON with "resourceLogs" { - PayloadCollector collector; + std::atomic count{0}; + std::vector payloads; logit::OtlpPayloadLogger::Config config; config.async = false; config.format.service_name = "sync-test"; - config.on_payload = [&collector](std::string payload) { - std::lock_guard lock(collector.mutex); - collector.payloads.push_back(std::move(payload)); - collector.count.fetch_add(1); - collector.cv.notify_all(); + config.on_payload = [&count, &payloads](std::string payload) { + payloads.push_back(std::move(payload)); + ++count; }; - LOGIT_ADD_LOGGER( - logit::OtlpPayloadLogger, - (config), - logit::SimpleLogFormatter, - ("%v") - ); + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); - LOGIT_WARN("sync payload test"); + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL, + "test.cpp", 10, "test_func", "sync payload test", "", + -1, false, false, false); + logger->log(record, "sync payload test"); - assert(collector.count.load() == 1); - assert(!collector.payloads.empty()); - assert(collector.payloads[0].find("\"resourceLogs\"") != std::string::npos); + assert(count.load() == 1); + assert(!payloads.empty()); + assert(payloads[0].find("\"resourceLogs\"") != std::string::npos); - LOGIT_SHUTDOWN(); + logger->shutdown(); } // Test a2: sync mode throwing callback increments failed exports @@ -65,25 +64,24 @@ int main() { throw std::runtime_error("sync payload rejected"); }; - LOGIT_ADD_LOGGER( - logit::OtlpPayloadLogger, - (config), - logit::SimpleLogFormatter, - ("%v") - ); + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); - LOGIT_WARN("sync throw test"); + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL, + "test.cpp", 20, "test_func", "sync throw test", "", + -1, false, false, false); + logger->log(record, "sync throw test"); assert(call_count.load() == 1); - uint64_t failed = static_cast(LOGIT_GET_INT_PARAM(0, logit::LoggerParam::FailedExportCount)); + uint64_t failed = static_cast(logger->get_int_param(logit::LoggerParam::FailedExportCount)); assert(failed >= 1); - LOGIT_SHUTDOWN(); + logger->shutdown(); } // Test b: async mode with batching - log 5 messages, verify callback receives 1 call with all 5 bodies { - PayloadCollector collector; + Collector collector; logit::OtlpPayloadLogger::Config config; config.async = true; @@ -97,20 +95,17 @@ int main() { collector.cv.notify_all(); }; - LOGIT_ADD_LOGGER( - logit::OtlpPayloadLogger, - (config), - logit::SimpleLogFormatter, - ("%v") - ); + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); - LOGIT_WARN("batch msg 1"); - LOGIT_WARN("batch msg 2"); - LOGIT_WARN("batch msg 3"); - LOGIT_WARN("batch msg 4"); - LOGIT_WARN("batch msg 5"); + for (int i = 1; i <= 5; ++i) { + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL + i, + "test.cpp", 30 + i, "test_func", "batch msg", "", + -1, false, false, false); + logger->log(record, "batch msg " + std::to_string(i)); + } - LOGIT_WAIT(); + logger->wait(); { std::unique_lock lock(collector.mutex); @@ -131,12 +126,12 @@ int main() { } assert(body_count == 5); - LOGIT_SHUTDOWN(); + logger->shutdown(); } // Test c: async mode queue overflow with drop_on_overflow=true { - PayloadCollector collector; + Collector collector; logit::OtlpPayloadLogger::Config config; config.async = true; @@ -152,28 +147,27 @@ int main() { collector.cv.notify_all(); }; - LOGIT_ADD_LOGGER( - logit::OtlpPayloadLogger, - (config), - logit::SimpleLogFormatter, - ("%v") - ); + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); for (int i = 0; i < 100; ++i) { - LOGIT_WARN("overflow msg"); + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL, + "test.cpp", 40, "test_func", "overflow msg", "", + -1, false, false, false); + logger->log(record, "overflow msg"); } std::this_thread::sleep_for(std::chrono::milliseconds(200)); - uint64_t dropped = static_cast(LOGIT_GET_INT_PARAM(0, logit::LoggerParam::DroppedLogCount)); + uint64_t dropped = static_cast(logger->get_int_param(logit::LoggerParam::DroppedLogCount)); assert(dropped > 0); - LOGIT_SHUTDOWN(); + logger->shutdown(); } // Test d: wait() blocks until queue drain { - PayloadCollector collector; + Collector collector; logit::OtlpPayloadLogger::Config config; config.async = true; @@ -187,15 +181,15 @@ int main() { collector.cv.notify_all(); }; - LOGIT_ADD_LOGGER( - logit::OtlpPayloadLogger, - (config), - logit::SimpleLogFormatter, - ("%v") - ); + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); - LOGIT_WARN("wait test message"); - LOGIT_WAIT(); + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL, + "test.cpp", 50, "test_func", "wait test message", "", + -1, false, false, false); + logger->log(record, "wait test message"); + + logger->wait(); { std::unique_lock lock(collector.mutex); @@ -206,12 +200,12 @@ int main() { assert(collector.count.load() >= 1); - LOGIT_SHUTDOWN(); + logger->shutdown(); } // Test e: shutdown() stops worker cleanly without deadlocks { - PayloadCollector collector; + Collector collector; logit::OtlpPayloadLogger::Config config; config.async = true; @@ -225,28 +219,25 @@ int main() { collector.cv.notify_all(); }; - LOGIT_ADD_LOGGER( - logit::OtlpPayloadLogger, - (config), - logit::SimpleLogFormatter, - ("%v") - ); + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); - LOGIT_WARN("shutdown test message"); + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL, + "test.cpp", 60, "test_func", "shutdown test message", "", + -1, false, false, false); + logger->log(record, "shutdown test message"); auto start = std::chrono::steady_clock::now(); - LOGIT_SHUTDOWN(); + logger->shutdown(); auto elapsed = std::chrono::duration_cast( std::chrono::steady_clock::now() - start).count(); assert(elapsed < 5000); - - LOGIT_SHUTDOWN(); } // Test f: wait() blocks until slow callback finishes { - PayloadCollector collector; + Collector collector; logit::OtlpPayloadLogger::Config config; config.async = true; @@ -261,24 +252,23 @@ int main() { collector.cv.notify_all(); }; - LOGIT_ADD_LOGGER( - logit::OtlpPayloadLogger, - (config), - logit::SimpleLogFormatter, - ("%v") - ); + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); - LOGIT_WARN("slow callback test"); + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL, + "test.cpp", 70, "test_func", "slow callback test", "", + -1, false, false, false); + logger->log(record, "slow callback test"); auto start = std::chrono::steady_clock::now(); - LOGIT_WAIT(); + logger->wait(); auto elapsed = std::chrono::duration_cast( std::chrono::steady_clock::now() - start).count(); assert(elapsed >= 400); assert(collector.count.load() >= 1); - LOGIT_SHUTDOWN(); + logger->shutdown(); } // Test g: throwing callback increments failed export count @@ -295,26 +285,115 @@ int main() { throw std::runtime_error("payload rejected"); }; - LOGIT_ADD_LOGGER( - logit::OtlpPayloadLogger, - (config), - logit::SimpleLogFormatter, - ("%v") - ); + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); - LOGIT_WARN("throw test 1"); - LOGIT_WARN("throw test 2"); + for (int i = 1; i <= 2; ++i) { + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL + i, + "test.cpp", 80 + i, "test_func", "throw test", "", + -1, false, false, false); + logger->log(record, "throw test " + std::to_string(i)); + } - LOGIT_WAIT(); - LOGIT_SHUTDOWN(); + logger->wait(); + logger->shutdown(); - // both logs should have been attempted (maybe in one batch, maybe two) assert(call_count.load() >= 1); - uint64_t failed = static_cast(LOGIT_GET_INT_PARAM(0, logit::LoggerParam::FailedExportCount)); + uint64_t failed = static_cast(logger->get_int_param(logit::LoggerParam::FailedExportCount)); assert(failed >= static_cast(call_count.load())); } + // Test h: async mode with payload splitting - small max_payload_bytes forces multiple chunks + { + Collector collector; + + logit::OtlpPayloadLogger::Config config; + config.async = true; + config.format.service_name = "async-split-test"; + config.max_batch_size = 256; + config.max_payload_bytes = 1024; + config.export_interval_ms = 50; + config.on_payload = [&collector](std::string payload) { + std::lock_guard lock(collector.mutex); + collector.payloads.push_back(std::move(payload)); + collector.count.fetch_add(1); + collector.cv.notify_all(); + }; + + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); + + for (int i = 0; i < 50; ++i) { + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL + i, + "test.cpp", 90 + i, "test_func", "async split test", "", + -1, false, false, false); + logger->log(record, "async split payload test message number " + std::to_string(i)); + } + + logger->wait(); + + { + std::unique_lock lock(collector.mutex); + collector.cv.wait_for(lock, std::chrono::seconds(3), [&collector]() { + return collector.count.load() >= 2; + }); + } + + assert(collector.count.load() > 1); + + int body_count = 0; + for (const auto& p : collector.payloads) { + std::size_t pos = 0; + while ((pos = p.find("\"body\"", pos)) != std::string::npos) { + ++body_count; + ++pos; + } + } + assert(body_count == 50); + + logger->shutdown(); + } + + // Test i: sync mode with small max_payload_bytes still exports all records + { + std::atomic count{0}; + std::vector payloads; + + logit::OtlpPayloadLogger::Config config; + config.async = false; + config.format.service_name = "sync-split-test"; + config.max_payload_bytes = 1024; + config.on_payload = [&count, &payloads](std::string payload) { + payloads.push_back(std::move(payload)); + ++count; + }; + + auto logger = std::unique_ptr(new logit::OtlpPayloadLogger(config)); + + for (int i = 0; i < 50; ++i) { + logit::LogRecord record( + logit::LogLevel::LOG_LVL_WARN, 1710000000123LL + i, + "test.cpp", 150 + i, "test_func", "sync split test", "", + -1, false, false, false); + logger->log(record, "sync split payload test message number " + std::to_string(i)); + } + + assert(count.load() > 1); + + int body_count = 0; + for (const auto& p : payloads) { + std::size_t pos = 0; + while ((pos = p.find("\"body\"", pos)) != std::string::npos) { + ++body_count; + ++pos; + } + } + assert(body_count == 50); + + logger->shutdown(); + } + return 0; } diff --git a/tests/otlp_payload_splitter_test.cpp b/tests/otlp_payload_splitter_test.cpp new file mode 100644 index 0000000..f66a1cb --- /dev/null +++ b/tests/otlp_payload_splitter_test.cpp @@ -0,0 +1,114 @@ +#include + +#ifdef LOGIT_WITH_OTLP + +#include +#include +#include +#include +#include + +namespace { + +logit::OtlpLogItem make_item(const std::string& message) { + logit::OtlpLogItem item; + item.record.log_level = logit::LogLevel::LOG_LVL_INFO; + item.record.timestamp_ms = 1710000000123LL; + item.record.file = "test.cpp"; + item.record.line = 1; + item.record.function = "test_func"; + item.record.format = "%v"; + item.record.thread_id = "t1"; + item.record.logger_index = -1; + item.record.print_mode = false; + item.record.fmt_mode = false; + item.record.raw_mode = false; + item.message = message; + return item; +} + +} // namespace + +int main() { + logit::OtlpJsonFormatConfig config; + config.service_name = "splitter-test"; + + // Test a: max_payload_bytes == 0 => single chunk for 10-record batch + { + std::vector batch; + for (int i = 0; i < 10; ++i) { + batch.push_back(make_item("msg " + std::to_string(i))); + } + + auto chunks = logit::build_otlp_logs_json_payload_chunks(batch, config, 0); + assert(chunks.size() == 1); + assert(chunks[0].find("\"resourceLogs\"") != std::string::npos); + } + + // Test b: small limit (1024 bytes) => batch of 50 records with long messages + // splits into >1 chunk; verify each chunk size <= 1024 OR oversized + { + std::vector batch; + for (int i = 0; i < 50; ++i) { + batch.push_back(make_item("payload-split-test-message-" + std::to_string(i))); + } + + auto chunks = logit::build_otlp_logs_json_payload_chunks(batch, config, 1024); + assert(chunks.size() > 1); + + // Each chunk must be <= 1024 bytes OR it's an oversized single-record chunk + for (const auto& chunk : chunks) { + // Count "body" occurrences to know how many records are in this chunk + std::size_t body_count = 0; + std::size_t pos = 0; + while ((pos = chunk.find("\"body\"", pos)) != std::string::npos) { + ++body_count; + ++pos; + } + if (body_count == 1) { + // Oversized single-record chunk is allowed + } else { + assert(chunk.size() <= 1024); + } + } + + // All 50 records must appear across all chunks + int total_bodies = 0; + for (const auto& chunk : chunks) { + std::size_t pos = 0; + while ((pos = chunk.find("\"body\"", pos)) != std::string::npos) { + ++total_bodies; + ++pos; + } + } + assert(total_bodies == 50); + } + + // Test c: empty batch => empty chunks vector + { + std::vector batch; + auto chunks = logit::build_otlp_logs_json_payload_chunks(batch, config, 1024); + assert(chunks.empty()); + } + + // Test d: single oversized record (5000-char message, limit 1024) => 1 chunk emitted + { + std::vector batch; + batch.push_back(make_item(std::string(5000, 'X'))); + + auto chunks = logit::build_otlp_logs_json_payload_chunks(batch, config, 1024); + assert(chunks.size() == 1); + assert(chunks[0].size() > 1024); + assert(chunks[0].find("\"resourceLogs\"") != std::string::npos); + } + + return 0; +} + +#else + +int main() { + return 0; +} + +#endif