Protect against concurrent reads/writes to Context keyvalues

benchmarks/benchmarks-core/src/jmh/java/io/micrometer/benchmark/core/ObservationKeyValuesBenchmark.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2024 VMware, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.micrometer.benchmark.core;
+
+import io.micrometer.common.KeyValues;
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+import java.util.concurrent.TimeUnit;
+
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Threads(4)
+public class ObservationKeyValuesBenchmark {
+
+    private static final KeyValues KEY_VALUES = KeyValues.of("key1", "value1", "key2", "value2", "key3", "value3",
+            "key4", "value4", "key5", "value5");
+
+    private final ObservationRegistry registry = ObservationRegistry.create();
+
+    private final Observation.Context context = new TestContext();
+
+    private final Observation observation = Observation.createNotStarted("jmh", () -> context, registry);
+
+    @Benchmark
+    @Group("contended")
+    @GroupThreads(1)
+    public Observation contendedWrite() {
+        return write();
+    }
+
+    @Benchmark
+    @Group("contended")
+    @GroupThreads(1)
+    public KeyValues contendedRead() {
+        return read();
+    }
+
+    @Benchmark
+    @Threads(1)
+    public Observation uncontendedWrite() {
+        return write();
+    }
+
+    @Benchmark
+    @Threads(1)
+    public KeyValues uncontendedRead() {
+        return read();
+    }
+
+    private Observation write() {
+        return observation.lowCardinalityKeyValues(KEY_VALUES);
+    }
+
+    private KeyValues read() {
+        return observation.getContext().getLowCardinalityKeyValues();
+    }
+
+    public static void main(String[] args) throws RunnerException {
+        Options options = new OptionsBuilder().include(ObservationKeyValuesBenchmark.class.getSimpleName())
+            .warmupIterations(3)
+            .measurementIterations(5)
+            .mode(Mode.SampleTime)
+            .forks(1)
+            .build();
+
+        new Runner(options).run();
+    }
+
+    static class TestContext extends Observation.Context {
+
+    }
+
+}

concurrency-tests/build.gradle

@@ -5,7 +5,6 @@ plugins {
 dependencies {
     implementation project(":micrometer-core")
 //    implementation("io.micrometer:micrometer-core:1.12.4")
-    implementation project(":micrometer-test")
     implementation project(":micrometer-registry-prometheus")
     runtimeOnly(libs.logbackLatest)
 }

concurrency-tests/src/jcstress/java/io/micrometer/concurrencytests/MeterRegistryConcurrencyTest.java

@@ -15,7 +15,6 @@
  */
 package io.micrometer.concurrencytests;
 
-import io.micrometer.core.Issue;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
@@ -154,7 +153,7 @@ public void arbiter(LL_Result r) {
      * iteration could happen at the same time a new meter is being registered, thus added
      * to the preFilterIdToMeterMap, modifying it while iterating over its KeySet.
      */
-    @Issue("gh-5489")
+    // @Issue("gh-5489")
     @JCStressTest
     @Outcome(id = "OK", expect = Expect.ACCEPTABLE, desc = "No exception")
     @Outcome(expect = Expect.FORBIDDEN, desc = "Exception thrown")

concurrency-tests/src/jcstress/java/io/micrometer/concurrencytests/ObservationContextConcurrencyTest.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2024 VMware, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.micrometer.concurrencytests;
+
+import io.micrometer.common.KeyValue;
+import io.micrometer.observation.Observation;
+import org.openjdk.jcstress.annotations.Actor;
+import org.openjdk.jcstress.annotations.JCStressTest;
+import org.openjdk.jcstress.annotations.Outcome;
+import org.openjdk.jcstress.annotations.State;
+import org.openjdk.jcstress.infra.results.LL_Result;
+
+import java.util.UUID;
+
+import static org.openjdk.jcstress.annotations.Expect.ACCEPTABLE;
+import static org.openjdk.jcstress.annotations.Expect.FORBIDDEN;
+
+public class ObservationContextConcurrencyTest {
+
+    @JCStressTest
+    @State
+    @Outcome(id = "No exception, No exception", expect = ACCEPTABLE)
+    @Outcome(expect = FORBIDDEN)
+    public static class ConsistentKeyValues {
+
+        private final Observation.Context context = new TestContext();
+
+        private final String uuid = UUID.randomUUID().toString();
+
+        @Actor
+        public void read(LL_Result r) {
+            try {
+                context.getHighCardinalityKeyValues();
+                r.r1 = "No exception";
+            }
+            catch (Exception e) {
+                r.r1 = e.getClass();
+            }
+        }
+
+        @Actor
+        public void write(LL_Result r) {
+            try {
+                context.addHighCardinalityKeyValue(KeyValue.of(uuid, uuid));
+                r.r2 = "No exception";
+            }
+            catch (Exception e) {
+                r.r2 = e.getClass();
+            }
+        }
+
+    }
+
+    static class TestContext extends Observation.Context {
+
+    }
+
+}

concurrency-tests/src/jcstress/java/io/micrometer/concurrencytests/PrometheusMeterRegistryConcurrencyTest.java

@@ -15,7 +15,6 @@
  */
 package io.micrometer.concurrencytests;
 
-import io.micrometer.core.Issue;
 import io.micrometer.core.instrument.DistributionSummary;
 import io.micrometer.core.instrument.LongTaskTimer;
 import io.micrometer.core.instrument.LongTaskTimer.Sample;
@@ -38,7 +37,7 @@
  */
 public class PrometheusMeterRegistryConcurrencyTest {
 
-    @Issue("#5193")
+    // @Issue("#5193")
     @JCStressTest
     @State
     @Outcome(id = "true", expect = ACCEPTABLE, desc = "Successful scrape")
@@ -67,7 +66,7 @@ public void scrape(Z_Result r) {
 
     }
 
-    @Issue("#5193")
+    // @Issue("#5193")
     @JCStressTest
     @State
     @Outcome(id = "true", expect = ACCEPTABLE, desc = "Successful scrape")
@@ -96,7 +95,7 @@ public void scrape(Z_Result r) {
 
     }
 
-    @Issue("#5193")
+    // @Issue("#5193")
     @JCStressTest
     @State
     @Outcome(id = "true", expect = ACCEPTABLE, desc = "Successful scrape")

micrometer-observation/src/main/java/io/micrometer/observation/Observation.java

@@ -21,7 +21,6 @@
 import io.micrometer.common.lang.Nullable;
 import io.micrometer.common.util.internal.logging.InternalLoggerFactory;
 
-import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
@@ -925,9 +924,9 @@ class Context implements ContextView {
         @Nullable
         private ObservationView parentObservation;
 
-        private final Map<String, KeyValue> lowCardinalityKeyValues = new LinkedHashMap<>();
+        private final Map<String, KeyValue> lowCardinalityKeyValues = new ConcurrentHashMap<>();
 
-        private final Map<String, KeyValue> highCardinalityKeyValues = new LinkedHashMap<>();
+        private final Map<String, KeyValue> highCardinalityKeyValues = new ConcurrentHashMap<>();
 
         /**
          * The observation name.