ELS publisher fails when publishing arrays with different types

[BrunoBonacci]

from Slack conversation https://clojurians.slack.com/archives/C03J782P329/p1737574229809269

from Nim Sadeh

Hey me again. Wondering if there are any tips for setting up structured logging with OpenSearch/ElasticSearch? I keep getting errors because of the dynamic nature of the structured logs. It's a bit unpredictable, so every day I squash a few, it works, then I get a new one.
For example, yesterday it complained about having periods in certain field names, so I put this dynamic template on:

{
  "dynamic_templates": [
    {
      "dots_as_keywords": {
        "path_match": ".*\\..*",
        "match_pattern": "regex",
        "mapping": {
          "type": "keyword"
        }
      }
    }
  ],
  "properties": {}
}

Today I am getting the error:

"illegal_argument_exception", :reason "mapper [info.o.trace.a] cannot be changed from type [long] to [text]"}}})}  

I'm wondering if there's a standard configuration that just works with mulog

[BrunoBonacci]

This is one of the pain of Elasticsearch/OpenSearch.
when you push data into Elasticsearch if you don’t have a defined schema it will infer one, that’s a great feature.
The inference will be based on the data has seen in that specific partition. So if you push a record like:

{ "foo": "bar", "qty": 3}

it will infer a schema that assumes records will have:

• a field called foo of type text
• a field called qty of type long

and this is great!!

Now if sometime late you post something like:

{ "foo": "bar", "qty": "unknown"}

it will try to index qty but in the previously assumed schema it was assumed to be a long and now this record contains "unknown" (a string).
This causes the type mismatch you have seen.

µ/log publisher for Elasticsearch has a feature called name mangling to work around this issue. You can read more about this: https://github.com/BrunoBonacci/mulog/blob/master/doc/els-name-mangling.md

in this case you have published an object like

{ "info": {"trace": [ 1 2 3 ]}}

and now you are trying to publish an object like

{ "info": {"trace": [ "foo" "bar"]}}

which for Elasticsearch inferred mapping is a type error.

I will post a workaround for this issue, and then improve the publisher's name mangling function to account for issues like that as well.

[BrunoBonacci]

The next version will have a fix for it,
meanwhile as a workaround you can use the following transformer in your publisher:

(require  '[clojure.walk :as w])

(defn mangle-arrays
  [events]
  (map
    (fn [event]
      (w/postwalk
        (fn [i]
          (if (or (set? i) (and (sequential? i) (not (map-entry? i))))
            (mapv (fn [item] {"_aVal" item}) i)
            i))
        event))
    events))

(μ/start-publisher!
  {:type :elasticsearch
   :url  "http://localhost:9200/"
   :transform mangle-arrays})

If you have existing data (and existing mapping) you might need to remove the existing mapping for that field. If you start fresh it should work.

Let me know if it works.

[nsadeh]

This is really awesome, thank you. The turnaround time is insane for an OS package, I would not expect something this fast from a paid service to be honest.

I wanted to ask about the trace stuff. You said:

in this case you have published an object like
{ "info": {"trace": [ 1 2 3 ]}}
and now you are trying to publish an object like

{ "info": {"trace": [ "foo" "bar"]}}
which for Elasticsearch inferred mapping is a type error.

I will post a workaround for this issue, and then improve the publisher's name mangling function to account for issues like that as well.

I am quite sure that this trace stuff comes from mulog itself. I have no objects where I published any float traces. I meanI have done a lot of mulog/trace ::my-keyword which I imagine is where this comes from, but I have not used an array in any of these (except for the arg array which in all of my traces is keywords) and no floats. This suggests to me that this is something that's generated automatically by mulog and therefore exists on all mulog installations, so it should have been a problem with OpenSearch before. Is that right, or have I just not looked deep enough at my codebase?

Here's an example trace in my codebase

(μ/trace ::generate-rag-prompt
                    [:phase :generate-rag-prompt]
                    (generate-rag-prompt past-messages))

(the redundant :phase is so that I can compare all phases).

All my traces look the same, because I only recently added traces to this codebase to analyze the phases of response generation for a complex RAG pipeline. Since they're all addressing only one use case they all have the same form.

[BrunoBonacci]

Hi thank you for the additional code.

the trace event from your code snippet will look like following:

{:mulog/event-name :your-ns/generate-rag-prompt,
 :mulog/timestamp  1587504242983,
 :mulog/trace-id   #mulog/flake "4VTF9QBbnef57vxVy-b4uKzh7dG7r7y4",
 :mulog/root-trace #mulog/flake "4VTF9QBbnef57vxVy-b4uKzh7dG7r7y4",
 :mulog/duration   254402837,
 :mulog/namespace  "your-ns",
 :mulog/outcome    :ok,
 :phase            :generate-rag-prompt,
}

which it doesn't explain the error.

I don't know where the error

"illegal_argument_exception", :reason "mapper [info.o.trace.a] cannot be changed from type [long] to [text]"}}})} 

came from, it might not be related to mulog at all.

[BrunoBonacci]

whether that particular illegal_argument_exception came from a log published by µ/log or not, the issue of publishing heterogeneous arrays was present.
The fix will be released with mulog-0.10.0

[nsadeh]

Not 100% sure yet that this error is caused by the transform proposed in this issue, but after implementing the proposed changes I am seeing the following error:

{
  :publisher-type :elasticsearch,
  :mulog/namespace "clojure.core",
  :publisher-id "50pXw9lTq1UcbB5V3a3758uL770v0j2W",
  :app-name "modeldiplomat-web-server",
  :mulog/action :publish,
  :mulog/timestamp 1737654087089,
  :exception #error {
    :cause "Elasticsearch Bulk API reported errors",
    :data {
      :errors ({
        :index {
          :_index "mulog-2025.01.23",
          :_id "50pY9T7qOwAP_y135jnUrPn91du_eJM7",
          :status 400,
          :error {
            :type "mapper_parsing_exception",
            :reason "failed to parse field [parsed.o.follow_ups.a] of type [text] in document with id '50pY9T7qOwAP_y135jnUrPn91du_eJM7'. Preview of field's value: '{_aVal.s=How is GIS currently utilized in the Amazon for land monitoring?}'",
            :caused_by {
              :type "illegal_state_exception",
              :reason "Can't get text on a START_OBJECT at 1:5313"
            }
          }
        }
      })
    },
    :via [
      {
        :type clojure.lang.ExceptionInfo,
        :message "Elasticsearch Bulk API reported errors",
        :data {
          :errors ({
            :index {
              :_index "mulog-2025.01.23",
              :_id "50pY9T7qOwAP_y135jnUrPn91du_eJM7",
              :status 400,
              :error {
                :type "mapper_parsing_exception",
                :reason "failed to parse field [parsed.o.follow_ups.a] of type [text] in document with id '50pY9T7qOwAP_y135jnUrPn91du_eJM7'. Preview of field's value: '{_aVal.s=How is GIS currently utilized in the Amazon for land monitoring?}'",
                :caused_by {
                  :type "illegal_state_exception",
                  :reason "Can't get text on a START_OBJECT at 1:5313"
                }
              }
            }
          })
        },
        :at [
          [com.brunobonacci.mulog.publishers.elasticsearch$post_records invokeStatic "elasticsearch.clj" 138]
        ]
      }
    ],
    :trace [
      [com.brunobonacci.mulog.publishers.elasticsearch$post_records invokeStatic "elasticsearch.clj" 138]
      [com.brunobonacci.mulog.publishers.elasticsearch$post_records invoke "elasticsearch.clj" 119]
      [com.brunobonacci.mulog.publishers.elasticsearch.ElasticsearchPublisher publish "elasticsearch.clj" 230]
      [com.brunobonacci.mulog.core$start_publisher_BANG_$publish_attempt__4134 invoke "core.clj" 194]
      [clojure.core$binding_conveyor_fn$fn__5823 invoke "core.clj" 2050]
      [clojure.lang.AFn applyToHelper "AFn.java" 154]
      [clojure.lang.RestFn applyTo "RestFn.java" 132]
      [clojure.lang.Agent$Action doRun "Agent.java" 114]
      [clojure.lang.Agent$Action run "Agent.java" 163]
      [java.util.concurrent.ThreadPoolExecutor runWorker "ThreadPoolExecutor.java" 1144]
      [java.util.concurrent.ThreadPoolExecutor$Worker run "ThreadPoolExecutor.java" 642]
      [java.lang.Thread run "Thread.java" 1583]
    ]
  },
  :version "0.1.0",
  :mulog/origin :mulog/core,
  :mulog/trace-id #mulog/flake "50pYAshkqzHMMITZzqIse2l2JPHKjSrV",
  :mulog/event-name :mulog/publisher-error
}

[BrunoBonacci]

As I mentioned earlier you will need to clear the existing mapping for that field
in this case the field is

{:parsed {:follow_ups [...] }}

it looks to me that somewhere in your code you log the output of the LLM and that structure has different/inconsistent shapes

in order for the workaround to be effective you will have to either

1. start on a fresh elasticsearch index
2. update the elasticsearch mapping for the troubled index and remove all the mappings for all the array values (https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-put-mapping.html)

[nsadeh]

I appreciate your patience, this is all new to me :)

[nsadeh]

BTW if you're curious the culprit is clojure.core/Throwable->map which generates a stacktrace with heterogenous data.

[BrunoBonacci]

That's strange as exception are converted to strings before being sent to Elasticsearch

[nsadeh]

I think it's no longer an exception at that point, just a map.

Would a better way of logging it out be to just do

(catch Throwable t
  (µ/log ::error
             :error-info t))

Instead of

(catch Throwable t
  (µ/log ::error
             :error-info (Throwable->map t)))