fix: make former queue thread safe.

inspectIT · Jul 19, 2024 · 028b9bc · 028b9bc
1 parent ab39da3
commit 028b9bc
Show file tree

Hide file tree

Showing 10 changed files with 83 additions and 69 deletions.
diff --git a/build.gradle b/build.gradle
@@ -118,6 +118,9 @@ dependencies {
     add("muzzleBootstrap", "io.opentelemetry.instrumentation:opentelemetry-instrumentation-annotations-support:${versions.opentelemetryJavaagentAlpha}")
     add("muzzleTooling", "io.opentelemetry.javaagent:opentelemetry-javaagent-extension-api:${versions.opentelemetryJavaagentAlpha}")
     add("muzzleTooling", "io.opentelemetry.javaagent:opentelemetry-javaagent-tooling:${versions.opentelemetryJavaagentAlpha}")
+
+    implementation 'com.github.ben-manes.caffeine:caffeine:3.1.8'
+
 }
 
 // Produces a copy of upstream javaagent with this extension jar included inside it

diff --git a/src/main/java/rocks/inspectit/gepard/agent/configuration/http/HttpConfigurationCallback.java b/src/main/java/rocks/inspectit/gepard/agent/configuration/http/HttpConfigurationCallback.java
@@ -34,7 +34,6 @@ public void completed(SimpleHttpResponse result) {
   // TODO Auslagern, vllt Util-Klasse?
   private InspectitConfiguration serializeConfiguration(String body) {
     ObjectMapper mapper = new ObjectMapper();
-    log.info(body);
     try {
       return mapper.readValue(body, InspectitConfiguration.class);
     } catch (IOException e) {

diff --git a/src/main/java/rocks/inspectit/gepard/agent/instrumentation/BatchInstrumenter.java b/src/main/java/rocks/inspectit/gepard/agent/instrumentation/BatchInstrumenter.java
@@ -21,26 +21,26 @@ public class BatchInstrumenter implements ClassDiscoveryListener, NamedRunnable
    */
   private final Instrumentation instrumentation;
 
-  private final ClassQueue classQueue;
+  private final InstrumentationCache instrumentationCache;
 
-  private BatchInstrumenter(ClassQueue classQueue) {
-    this.classQueue = classQueue;
+  private BatchInstrumenter(InstrumentationCache instrumentationCache) {
+    this.instrumentationCache = instrumentationCache;
     this.instrumentation = InstrumentationHolder.getInstrumentation();
   }
 
-  public static BatchInstrumenter create(ClassQueue classQueue) {
-    BatchInstrumenter instrumenter = new BatchInstrumenter(classQueue);
+  public static BatchInstrumenter create(InstrumentationCache instrumentationCache) {
+    BatchInstrumenter instrumenter = new BatchInstrumenter(instrumentationCache);
     return instrumenter;
   }
 
   @Override
   public void onNewClassesDiscovered(Set<Class<?>> newClasses) {
-    classQueue.addAll(newClasses);
+    instrumentationCache.addAll(newClasses);
   }
 
   @Override
   public void run() {
-    log.info("Retransforming classes...");
+    // log.info("Retransforming classes...");
     try{
       Set<Class<?>> batch = getBatch(BATCH_SIZE);
       Iterator<Class<?>> batchIterator = batch.iterator();
@@ -56,6 +56,7 @@ private void retransformBatch(Iterator<Class<?>> batchIterator) {
       Class<?> clazz = batchIterator.next();
       batchIterator.remove();
       try {
+        log.info("Retransforming class {}", clazz.getName());
         instrumentation.retransformClasses(clazz);
       } catch (Throwable e) {
         log.error("Error while retransforming class {}", clazz.getName(), e);
@@ -69,12 +70,14 @@ public String getName() {
   }
 
   public Set<Class<?>> getBatch(int batchSize) {
-    Set<Class<?>> batch = new HashSet<>();
-    Iterator<Class<?>> queueIterator = classQueue.getPendingClasses().iterator();
+    Set<Class<?>> batch = new HashSet<>();;
     int checkedClassesCount = 0;
+    Iterator<Class<?>> queueIterator = instrumentationCache.getKeyIterator();
+
     while (queueIterator.hasNext()) {
       Class<?> clazz = queueIterator.next();
       queueIterator.remove();
+      instrumentationCache.remove(clazz);
       checkedClassesCount++;
       if (ConfigurationResolver.shouldRetransform(clazz)) {
         batch.add(clazz);
@@ -85,12 +88,13 @@ public Set<Class<?>> getBatch(int batchSize) {
       }
     }
 
-    if (!batch.isEmpty()) {
-      log.info(
+
+
+      log.debug(
           "Checked configuration of {} classes, {} classes left to check",
           checkedClassesCount,
-          classQueue.getPendingClasses().size());
-    }
+          instrumentationCache.getSize());
+
     return batch;
   }
 }
diff --git a/src/main/java/rocks/inspectit/gepard/agent/instrumentation/ClassQueue.java b/src/main/java/rocks/inspectit/gepard/agent/instrumentation/ClassQueue.java
diff --git a/src/main/java/rocks/inspectit/gepard/agent/instrumentation/ConfigurationReceiver.java b/src/main/java/rocks/inspectit/gepard/agent/instrumentation/ConfigurationReceiver.java
@@ -4,6 +4,9 @@
 import java.lang.instrument.Instrumentation;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Map;
+import java.util.stream.Collectors;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import rocks.inspectit.gepard.agent.internal.configuration.observer.ConfigurationReceivedEvent;
@@ -14,17 +17,17 @@ public class ConfigurationReceiver implements ConfigurationReceivedObserver {
   private final Logger logger = LoggerFactory.getLogger(ConfigurationReceiver.class);
   private final Instrumentation instrumentation;
 
-  private final ClassQueue classQueue;
+  private final InstrumentationCache instrumentationCache;
 
-  public ConfigurationReceiver(ClassQueue classQueue) {
-    this.classQueue = classQueue;
+  public ConfigurationReceiver(InstrumentationCache instrumentationCache) {
+    this.instrumentationCache = instrumentationCache;
     this.instrumentation = InstrumentationHolder.getInstrumentation();
   }
 
   @Override
   public void handleConfiguration(ConfigurationReceivedEvent event) {
-    Collections.addAll(
-        Arrays.asList(classQueue.getPendingClasses().toArray()),
-        instrumentation.getAllLoadedClasses());
+    // Make Map from instrumentation.getAllLoadedClasses()
+    Class<?>[] loadedClasses = instrumentation.getAllLoadedClasses();
+    instrumentationCache.addAll(Arrays.stream(loadedClasses).collect(Collectors.toSet()));
   }
 }
diff --git a/src/main/java/rocks/inspectit/gepard/agent/instrumentation/InstrumentationCache.java b/src/main/java/rocks/inspectit/gepard/agent/instrumentation/InstrumentationCache.java
@@ -0,0 +1,36 @@
+package rocks.inspectit.gepard.agent.instrumentation;
+
+
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+
+import java.util.*;
+
+//TODO: Make Class Queue Thread SAFE!
+public class InstrumentationCache {
+  private final Cache<Class<?>, Boolean> pendingClasses;
+
+  public InstrumentationCache() {
+    this.pendingClasses = Caffeine.newBuilder()
+            .build();
+  }
+
+    public void addAll(Set<Class<?>> classes) {
+        for (Class<?> clazz : classes) {
+            pendingClasses.put(clazz, true);
+        }
+    }
+
+    public Iterator<Class<?>> getKeyIterator() {
+        return pendingClasses.asMap().keySet().iterator();
+    }
+
+    public long getSize() {
+        return pendingClasses.estimatedSize();
+    }
+
+    public void remove(Class<?> clazz) {
+        pendingClasses.invalidate(clazz);
+    }
+}
diff --git a/src/main/java/rocks/inspectit/gepard/agent/instrumentation/InstrumentationManager.java b/src/main/java/rocks/inspectit/gepard/agent/instrumentation/InstrumentationManager.java
@@ -9,33 +9,33 @@
 public class InstrumentationManager {
   private static final Logger log = LoggerFactory.getLogger(InstrumentationManager.class);
 
-  private final ClassQueue classQueue;
+  private final InstrumentationCache instrumentationCache;
 
-  private InstrumentationManager(ClassQueue classQueue) {
-    this.classQueue = classQueue;
+  private InstrumentationManager(InstrumentationCache instrumentationCache) {
+    this.instrumentationCache = instrumentationCache;
   }
 
   public static InstrumentationManager create() {
-    ClassQueue classQueue = new ClassQueue();
-    return new InstrumentationManager(classQueue);
+    InstrumentationCache instrumentationCache = new InstrumentationCache();
+    return new InstrumentationManager(instrumentationCache);
   }
 
   public void startClassDiscovery() {
     InspectitScheduler scheduler = InspectitScheduler.getInstance();
-    ClassDiscoveryService discoveryService = new ClassDiscoveryService(classQueue);
+    ClassDiscoveryService discoveryService = new ClassDiscoveryService(instrumentationCache);
     Duration discoveryInterval = Duration.ofSeconds(10);
     scheduler.startRunnable(discoveryService, discoveryInterval);
   }
 
   public void startBatchInstrumentation() {
     InspectitScheduler scheduler = InspectitScheduler.getInstance();
-    BatchInstrumenter batchInstrumenter = BatchInstrumenter.create(classQueue);
+    BatchInstrumenter batchInstrumenter = BatchInstrumenter.create(instrumentationCache);
     Duration batchInterval = Duration.ofMillis(500);
     scheduler.startRunnable(batchInstrumenter, batchInterval);
   }
 
   public void createConfigurationReceiver() {
-    ConfigurationReceiver configurationReceiver = new ConfigurationReceiver(classQueue);
+    ConfigurationReceiver configurationReceiver = new ConfigurationReceiver(instrumentationCache);
     configurationReceiver.subscribeToConfigurationReceivedEvents();
   }
 }
diff --git a/...in/java/rocks/inspectit/gepard/agent/instrumentation/discovery/ClassDiscoveryService.java b/...in/java/rocks/inspectit/gepard/agent/instrumentation/discovery/ClassDiscoveryService.java
@@ -8,7 +8,7 @@
 import java.util.WeakHashMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import rocks.inspectit.gepard.agent.instrumentation.ClassQueue;
+import rocks.inspectit.gepard.agent.instrumentation.InstrumentationCache;
 import rocks.inspectit.gepard.agent.internal.schedule.NamedRunnable;
 
 public class ClassDiscoveryService implements NamedRunnable {
@@ -18,11 +18,11 @@ public class ClassDiscoveryService implements NamedRunnable {
 
   private final Instrumentation instrumentation;
 
-  private final ClassQueue classQueue;
+  private final InstrumentationCache instrumentationCache;
 
-  public ClassDiscoveryService(ClassQueue classQueue) {
+  public ClassDiscoveryService(InstrumentationCache instrumentationCache) {
     this.instrumentation = InstrumentationHolder.getInstrumentation();
-    this.classQueue = classQueue;
+    this.instrumentationCache = instrumentationCache;
   }
 
   @Override
@@ -49,7 +49,7 @@ void discoverClasses() {
       }
     }
     log.debug("Discovered {} new classes", newClasses.size());
-    classQueue.addAll(newClasses);
+    instrumentationCache.addAll(newClasses);
   }
 
   @Override

diff --git a/src/main/java/rocks/inspectit/gepard/agent/transformation/DynamicTransformer.java b/src/main/java/rocks/inspectit/gepard/agent/transformation/DynamicTransformer.java
@@ -24,7 +24,7 @@ public DynamicType.Builder<?> transform(
       JavaModule module,
       ProtectionDomain protectionDomain) {
     if (ConfigurationResolver.shouldInstrument(typeDescription)) {
-      log.debug("Transforming type: {}", typeDescription.getName());
+      log.info("Transforming type: {}", typeDescription.getName());
 
       builder = builder.visit(Advice.to(InspectitAdvice.class).on(any()));
     }

diff --git a/start-example-app.sh b/start-example-app.sh
@@ -1,9 +1,11 @@
 #!/bin/bash
-  java -agentlib:jdwp="transport=dt_socket,server=y,suspend=y,address=127.0.0.1:8000" \
+  java  \
         -javaagent:build/libs/opentelemetry-javaagent.jar \
         -Dotel.service.name="my-service" \
         -Dinspectit.config.http.url="https://localhost:8080/api/v1" \
         -Djavax.net.ssl.trustStore="src/main/resources/keystore/inspectit-dev-keystore.jks" \
         -Djavax.net.ssl.trustStorePassword="Ocecat24" \
         -Dotel.javaagent.configuration-file=otel-configuration.properties \
-        -jar /Users/lkr/Documents/Projects/VHV/02_ocelot/java-21-rest-example/build/libs/java-21-rest-example-0.0.1-SNAPSHOT.jar
+        -jar /Users/lkr/Documents/Projects/VHV/02_ocelot/java-21-rest-example/build/libs/java-21-rest-example-0.0.1-SNAPSHOT.jar
+
+#-agentlib:jdwp="transport=dt_socket,server=y,suspend=y,address=127.0.0.1:8000"\