[BUG] The user_agent processor throws exceptions with multiple threads.

[dlvenable]

Describe the bug

I'm seeing a number of errors such as:

2024-06-05T20:54:54.420 [my-pipeline-processor-worker-1-thread-2] ERROR org.opensearch.dataprepper.plugins.processor.useragent.UserAgentProcessor - An exception occurred when parsing user agent data from event [******] with source key [******]
java.lang.IllegalStateException: Entry.next=null, data[removeIndex]=***
 previous=***
 key=***
 size=1000 maxSize=1000 This should not occur if your keys are immutable, and you have used synchronization properly.
 at org.apache.commons.collections4.map.LRUMap.reuseMapping(LRUMap.java:384) ~[commons-collections4-4.4.jar:4.4]
 at org.apache.commons.collections4.map.LRUMap.addMapping(LRUMap.java:349) ~[commons-collections4-4.4.jar:4.4]
 at org.apache.commons.collections4.map.AbstractHashedMap.put(AbstractHashedMap.java:289) ~[commons-collections4-4.4.jar:4.4]
 at ua_parser.CachingParser.parseUserAgent(CachingParser.java:92) ~[uap-java-1.6.1.jar:?]
 at ua_parser.Parser.parse(Parser.java:83) ~[uap-java-1.6.1.jar:?]
 at ua_parser.CachingParser.parse(CachingParser.java:72) ~[uap-java-1.6.1.jar:?]
 at org.opensearch.dataprepper.plugins.processor.useragent.UserAgentProcessor.doExecute(UserAgentProcessor.java:51) ~[user-agent-processor-2.x.88.jar:?]
 at org.opensearch.dataprepper.model.processor.AbstractProcessor.lambda$execute$0(AbstractProcessor.java:54) ~[data-prepper-api-2.x.88.jar:?]
 at io.micrometer.core.instrument.composite.CompositeTimer.record(CompositeTimer.java:69) ~[micrometer-core-1.13.0.jar:1.13.0]
 at org.opensearch.dataprepper.model.processor.AbstractProcessor.execute(AbstractProcessor.java:54) ~[data-prepper-api-2.x.88.jar:?]

To Reproduce

It is not very easy to reproduce this consistently.

1. Create a pipeline with the user_agent processor.
2. Be sure to have multiple workers in the pipeline.
3. Ingest a significant volume of data
4. These errors should occur.

Expected behavior

No errors, and cache works.

[dlvenable]

I found this issue which seems to be the cause.

ua-parser/uap-java#41

Rather than try to synchronize in the processor, I'm going to write a new implementation of Parser which uses a Caffeine cache.

[dlvenable]

Reproduction

pipeline.yaml

empty-dlq-pipeline:
  workers: 2
  source:
    http:

  buffer:
    bounded_blocking:
      buffer_size: 500000
      batch_size: 10000

  processor:
    - user_agent:
        source: log
        target: user_agent

  sink:
    #- stdout:
    - opensearch:
        hosts: [ "https://opensearch:9200" ]
        insecure: true
        username: admin
        password: admin
        index: test_documents

docker-compose.yml

services:
  data-prepper:
    container_name: data-prepper
    #image: opensearchproject/data-prepper:2
    image: opensearch-data-prepper:2.9.0-SNAPSHOT
    depends_on:
      - opensearch
    volumes:
      - ./data-prepper/pipelines:/usr/share/data-prepper/pipelines
      - ./data-prepper/config/data-prepper-config.yaml:/usr/share/data-prepper/config/data-prepper-config.yaml
    ports:
      - "2021:2021"
      - "4900:4900"
    networks:
      - data-prepper-test

  opensearch:
    container_name: opensearch
    image: opensearchproject/opensearch:2.11.1
    environment:
      - discovery.type=single-node
      - bootstrap.memory_lock=true # along with the memlock settings below, disables swapping
      - "OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m" # minimum and maximum Java heap size, recommend setting both to 50% of system RAM
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536 # maximum number of open files for the OpenSearch user, set to at least 65536 on modern systems
        hard: 65536
    ports:
      - 9200:9200
      - 9600:9600 # required for Performance Analyzer
    networks:
      - data-prepper-test
  dashboards:
    image: opensearchproject/opensearch-dashboards:2.11.1
    container_name: opensearch-dashboards
    ports:
      - 5601:5601
    expose:
      - "5601"
    environment:
      OPENSEARCH_HOSTS: '["https://opensearch:9200"]'
    depends_on:
      - opensearch
    networks:
      - data-prepper-test

networks:
  data-prepper-test:

Test execution

Change Chain::sendApacheCommonLogPostRequest in performance-test project to:

    public static ChainBuilder sendApacheCommonLogPostRequest(final String name, final int batchSize) {
        return CoreDsl.exec(
                HttpDsl.http(name)
                        .post(PathTarget.getPath())
                        .body(CoreDsl.StringBody(Templates.userAgent(batchSize))));
    }

Run:

./gradlew :performance-test:gatlingRun-org.opensearch.dataprepper.test.performance.TargetRpsSimulation