processor_fts: add JSON field decoding and log body unpacking

plugins/processor_fts/README.md

@@ -184,40 +184,82 @@ order is preserved.
 | `fb.set_severity(log, level)` | Set severity text and number from a level name: `trace`, `debug`, `info`, `warn`, `error`, `fatal`, and common aliases. |
 | `fb.set_trace_context(log, trace_hex, span_hex)` | Set the trace and span IDs from hexadecimal strings. IDs must be nonzero. |
 | `fb.log_field(log, name)` | Read an OpenTelemetry log field such as `severity_number`, `trace_id`, or `observed_timestamp`. |
+| `fb.unpack_json(log, field)` | Decode a JSON object from a body field (`message` by default) and replace `log.body`. Return `true` on success or `false, error` for a missing/non-string field, ordinary parse failure, or non-object JSON; the original body is preserved for these failures. Uses the same typed parser and limits as `fb.parse_json`. |
+
+Instruction-limit exhaustion in the JSON helpers fails the callback instead of
+returning a recoverable decode error. Changes are rolled
+back and `error_mode` determines the record outcome: the default `drop` removes
+the record, while `keep` forwards the original.
 
 ### Example: parse an embedded JSON message
 
 ```lua
 local fb = require("fluentbit")
 
 function process(log)
-    local text = log.body:get_string("message")
-    if not text or not text:match("^%s*{") then
-        return
-    end
+    fb.unpack_json(log, "message")
+    fb.redact(log.body, {"authorization", "password"})
+end
+```
 
-    local ok, app = pcall(fb.parse_json, text)
-    if not ok then
-        return fb.DROP
-    end
+The example allows mixed streams to pass plain text through. The unpack call
+keeps plain text, missing/non-string fields, malformed JSON, and non-object JSON
+unchanged. `fb.redact` matches keys in maps and arrays; it does not inspect
+secrets inside the unchanged message string. If every forwarded message must be
+decoded for key-based redaction, drop records when decoding fails:
 
-    fb.redact(app, {"authorization", "password"})
-    log.body = app
+```lua
+function process(log)
+    if not fb.unpack_json(log, "message") then return fb.DROP end
+    fb.redact(log.body, {"authorization", "password"})
+end
+```
+
+### Decode a field in place
+
+Use `fb.decode_json(container, key)` to replace only the selected string field
+with its decoded value. The same helper works with body maps, nested maps,
+OpenTelemetry attributes, and array elements:
 
+```lua
+function process(log)
+    fb.decode_json(log.body, "message")
     if fb.is_otel(log) then
-        local level = app:get_string("level")
-        if level then
-            fb.set_severity(log, level)
-        end
-
-        local time = app:get_string("time")
-        if time then
-            log.timestamp = fb.parse_time(time)
-        end
+        fb.decode_json(log.attributes, "payload")
+        fb.decode_json(log.resource.attributes, "labels")
     end
+    fb.redact(log.body, {"authorization", "password"})
 end
 ```
 
+For example, `{"message":"{\"level\":\"info\"}","host":"web"}` becomes
+`{"message":{"level":"info"},"host":"web"}`. By comparison,
+`fb.unpack_json(log, "message")` produces `{"level":"info"}` as the whole body.
+Both operations decode one selected string; strings inside the decoded JSON
+require another explicit call.
+
+`fb.decode_json` accepts any valid JSON value, including arrays, null, and
+booleans. `fb.unpack_json` requires an object because a log body must be a map.
+Missing/non-string fields and ordinary parse failures return `false, error`
+without changing the target. Successful decoding returns `true`. Invalid
+container/key arguments, expired views, and failures while applying a mutation
+raise errors and follow the callback's `error_mode`.
+
+For OpenTelemetry log bodies and attributes, the JSON helpers reject unsigned
+integers above `9223372036854775807` (`INT64_MAX`), including values nested in
+objects or arrays. OTLP AnyValue integers must fit signed 64 bits. The helpers
+return `false, error` before mutation, preserving the original field or body.
+Ordinary logs, detached scratch maps, `fb.parse_json`, and `fb.try_parse_json`
+retain full unsigned 64-bit support.
+
+These helper statuses are separate from callback results. Leaving `process`
+without an explicit return is equivalent to `return fb.KEEP`: successful
+changes are committed and the record is forwarded. To remove a record, return
+`fb.DROP` explicitly; do not return a helper's boolean status from `process`.
+
+Use `fb.try_parse_json(text)` when you need the parsed value without mutating a
+container. `fb.parse_json(text)` remains the throwing parser.
+
 ## Metrics
 
 `process(metric)` runs once per metric family. Resource and scope changes apply
@@ -316,6 +358,8 @@ parent references.
 | `fb.redact(value, keys, replacement)` | Replace the values of matching keys in nested maps and arrays. Key matching is case-insensitive. The default replacement is `[REDACTED]`. |
 | `fb.keep_keys(map, keys)` | Remove every key not listed. |
 | `fb.parse_json(text)` | Parse a JSON document into typed values. Integer widths are preserved; duplicate keys and nesting beyond 64 levels are rejected. Input is limited to 1 MiB. |
+| `fb.decode_json(container, key)` | Decode a string field in a typed map or array in place. Return `true` on success or `false, error` for a missing/non-string field or ordinary parse failure. Preserve sibling fields and leave the source unchanged on failure. Array indexes start at 1; map keys are literal. |
+| `fb.try_parse_json(text)` | Return the same typed value as `fb.parse_json`, or `nil, error` for invalid JSON or non-string input (including a missing field). JSON `null` returns a typed null value, and JSON `false` returns a typed boolean value. The same parser and policy limits apply. |
 | `fb.parse_time(text)` | Parse an RFC 3339 timestamp into `{sec, nsec}`. |
 
 Key-based redaction does not inspect free-form strings; secrets embedded in

plugins/processor_fts/fts.h

@@ -156,6 +156,8 @@ struct fts_runtime {
     int empty_metatable_ref;
     int key_refs[FTS_KEY_COUNT];
     int tag_ref;
+    /* Current OTLP event for destination-aware JSON validation. */
+    int json_event_ref;
     uint64_t generation;
     struct fts_undo_entry *undo;
     size_t undo_count;

plugins/processor_fts/fts_log_helpers.lua

@@ -1,4 +1,18 @@
 local fb = package.loaded.fluentbit
+--- Replace a log body with a decoded object; preserve it and return false, error on failure.
+function fb.unpack_json(log, field)
+  local source = log.body:get(field or 'message')
+  if not source or source:type() ~= 'string' then
+    return false, 'JSON source field must be a string'
+  end
+  local value, err = fb.try_parse_json(source:value())
+  if not value then return false, err end
+  if value:type() ~= 'map' then return false, 'JSON body must be an object' end
+  err = fb._fts_json_integer_error(log.body, value)
+  if err then return false, err end
+  log.body = value
+  return true
+end
 function fb.is_otel(log)
   local schema = rawget(log, 'context') and rawget(log, 'context').metadata:get('schema')
   return schema and schema:type() == 'string' and schema:value() == 'otlp' or false

plugins/processor_fts/fts_logs.c

@@ -289,6 +289,120 @@ static struct cfl_variant *log_field(struct cfl_variant *metadata, const char *k
     return value ? value : map_field(metadata, key);
 }
 
+/* Compare live ancestors, including views into a replaced log body. */
+static int json_target_within(lua_State *l, struct cfl_variant *payload)
+{
+    struct fts_value *value;
+    int base;
+    int depth;
+
+    if (!payload) {
+        return FLB_FALSE;
+    }
+
+    base = lua_gettop(l);
+    lua_pushvalue(l, 1);
+    for (depth = 0; depth < FTS_MAX_DEPTH; depth++) {
+        if (fts_resolve_value(l, -1, 0) == payload) {
+            lua_settop(l, base);
+            return FLB_TRUE;
+        }
+        value = lua_touserdata(l, -1);
+        if (value->owned) {
+            break;
+        }
+        fts_lua_getuservalue(l, -1);
+        lua_rawgeti(l, -1, 1);
+        lua_replace(l, base + 1);
+        lua_settop(l, base + 1);
+    }
+    lua_settop(l, base);
+    return FLB_FALSE;
+}
+
+/* Read a typed field from an already validated table without invoking policy code. */
+static struct cfl_variant *json_payload_field(lua_State *l, int table, const char *key)
+{
+    struct cfl_variant *value;
+
+    lua_pushstring(l, key);
+    lua_rawget(l, table);
+    value = fts_resolve_value(l, -1, 0);
+    lua_pop(l, 1);
+    return value;
+}
+
+/* Check every nested integer while charging traversal to the policy budget. */
+static int json_integers_fit(lua_State *l, struct fts_runtime *ctx,
+                             struct cfl_variant *value, int depth)
+{
+    struct cfl_list *head;
+    struct cfl_kvpair *pair;
+    size_t i;
+
+    if (depth >= FTS_MAX_DEPTH || (!ctx->jit && --ctx->instructions_left <= 0)) {
+        return luaL_error(l, "JSON integer validation exceeds nesting or instruction limit");
+    }
+    if (value->type == CFL_VARIANT_UINT) {
+        return value->data.as_uint64 <= INT64_MAX;
+    }
+    if (value->type == CFL_VARIANT_KVLIST) {
+        cfl_list_foreach(head, &value->data.as_kvlist->list) {
+            pair = cfl_list_entry(head, struct cfl_kvpair, _head);
+            if (!json_integers_fit(l, ctx, pair->val, depth + 1)) {
+                return FLB_FALSE;
+            }
+        }
+    }
+    else if (value->type == CFL_VARIANT_ARRAY) {
+        for (i = 0; i < value->data.as_array->entry_count; i++) {
+            if (!json_integers_fit(l, ctx, value->data.as_array->entries[i], depth + 1)) {
+                return FLB_FALSE;
+            }
+        }
+    }
+    return FLB_TRUE;
+}
+
+/* JSON decode failures must be recoverable before OTLP commit validation. */
+static int json_integer_error(lua_State *l)
+{
+    struct fts_runtime *ctx;
+    struct cfl_variant *metadata;
+    struct cfl_variant *value;
+    int event;
+    int payload;
+
+    lua_getallocf(l, (void **) &ctx);
+    lua_rawgeti(l, LUA_REGISTRYINDEX, ctx->json_event_ref);
+    if (!lua_istable(l, -1)) {
+        lua_pushnil(l);
+        return 1;
+    }
+    event = lua_gettop(l);
+    payload = json_target_within(l, json_payload_field(l, event, "body"));
+    if (!payload) {
+        metadata = json_payload_field(l, event, "metadata");
+        payload = json_target_within(l, log_field(metadata, "attributes"));
+    }
+    if (!payload) {
+        lua_pushliteral(l, "context");
+        lua_rawget(l, event);
+        if (lua_istable(l, -1)) {
+            value = json_payload_field(l, lua_gettop(l), "attributes");
+            payload = json_target_within(l, value);
+        }
+        lua_pop(l, 1);
+    }
+    if (payload && !json_integers_fit(l, ctx, fts_resolve_value(l, 2, 0), 0)) {
+        lua_pushliteral(l, "OTel AnyValue integers must fit signed 64 bits");
+    }
+    else {
+        lua_pushnil(l);
+    }
+    return 1;
+}
+
 static void push_changed_fields(lua_State *l, struct record_call *call,
                                 struct cfl_variant *metadata, struct cfl_variant *context)
 {
@@ -578,6 +692,7 @@ static void require_plain_table(lua_State *l, int index, const char *name)
     }
 }
 
+/* Run one log policy with transactional values and validate its committed result. */
 static int run_record(lua_State *l)
 {
     struct record_call *call;
@@ -649,6 +764,8 @@ static int run_record(lua_State *l)
     }
 
     if (has_otel_context(call->metadata)) {
+        lua_pushvalue(l, event);
+        lua_rawseti(l, LUA_REGISTRYINDEX, call->ctx->json_event_ref);
         lua_rawgeti(l, LUA_REGISTRYINDEX, call->ctx->module_ref);
         lua_getfield(l, -1, "_attach");
         lua_pushvalue(l, event);
@@ -1976,8 +2093,22 @@ int fts_process_logs(struct flb_processor_instance *ins, void *data,
     return FLB_PROCESSOR_FAILURE;
 }
 
+/* Reserve callback context storage and expose native and Lua log helpers. */
 static int register_log_helpers(lua_State *l)
 {
+    struct fts_runtime *ctx;
+
+    lua_getallocf(l, (void **) &ctx);
+    lua_pushboolean(l, 0);
+    ctx->json_event_ref = luaL_ref(l, LUA_REGISTRYINDEX);
+
+    lua_getglobal(l, "package");
+    lua_getfield(l, -1, "loaded");
+    lua_getfield(l, -1, "fluentbit");
+    lua_pushcfunction(l, json_integer_error);
+    lua_setfield(l, -2, "_fts_json_integer_error");
+    lua_pop(l, 3);
+
     if (fts_load_helpers(l, fts_log_helpers, sizeof(fts_log_helpers) - 1,
                          "fluentbit.log_helpers") != 0) {
         return -1;

plugins/processor_fts/fts_runtime.c

@@ -363,12 +363,14 @@ static int initialize_policy(lua_State *l)
     return 0;
 }
 
+/* Initialize the bounded Lua runtime and the selected signal adapter's helpers. */
 int fts_runtime_initialize(struct fts_runtime *ctx, struct flb_config *config)
 {
     lua_State *l;
     int ret;
 
     ctx->env = config->env;
+    ctx->json_event_ref = LUA_NOREF;
     ctx->vm = flb_luajit_create(config);
     if (!ctx->vm) {
         return FLB_PROCESSOR_FAILURE;
@@ -451,6 +453,11 @@ int fts_runtime_call(struct fts_runtime *ctx, void *data, int results)
     if (ret != 0) {
         fts_sanitizer_forget_unwound_frames();
     }
+    /* Release callback-only roots before error-path garbage collection. */
+    if (ctx->json_event_ref != LUA_NOREF) {
+        lua_pushboolean(l, 0);
+        lua_rawseti(l, LUA_REGISTRYINDEX, ctx->json_event_ref);
+    }
     if (ret == 0 && !ctx->jit && ctx->instructions_left <= 0) {
         lua_settop(l, base);
         lua_pushliteral(l, "Lua instruction limit exceeded");

plugins/processor_fts/fts_value_helpers.lua

@@ -1,4 +1,25 @@
 local fb = package.loaded.fluentbit
+--- Parse JSON into a typed value, or return nil and the ordinary parse error.
+function fb.try_parse_json(text)
+  local ok, value = pcall(fb.parse_json, text)
+  if ok then return value end
+  return nil, value
+end
+--- Decode one string field in place; return false and an error without mutating on failure.
+function fb.decode_json(container, key)
+  local source = container:get(key)
+  if not source or source:type() ~= 'string' then
+    return false, 'JSON source field must be a string'
+  end
+  local value, err = fb.try_parse_json(source:value())
+  if not value then return false, err end
+  if fb._fts_json_integer_error then
+    err = fb._fts_json_integer_error(container, value)
+    if err then return false, err end
+  end
+  container:set(key, value)
+  return true
+end
 function fb.keep_keys(map, keys)
   local keep = {}; for _, k in ipairs(keys) do keep[k] = true end
   for _, k in ipairs(map:keys()) do if not keep[k] then map:remove(k) end end