Propagate context in queue

[harveychong]

I would like to trace the following code:

package com.hello;

import java.util.concurrent.ArrayBlockingQueue;

public class FacadeMain {

    public static void main(String[] args) throws InterruptedException {
        FacadeMain facadeMain = new FacadeMain();
        for (int i = 0; i < 10; i++) {
            facadeMain.putData(i);
        }
    }

    private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(20);

    public FacadeMain() {
        new Thread(() -> {
            int count = 0;
            while (true) {
                final Integer data = queue.poll();

                if (data != null) {
                    sendData(data);
                    count++;

                    if (count >= 10) {
                        break;
                    }
                }
            }

        }).start();
    }

    public void putData(Integer a) throws InterruptedException {
        queue.put(a);
    }

    public void sendData(Integer a) {
        System.out.println(Thread.currentThread().getName() + ": " + a);
    }
}

I went through the source code and there is no exitsing extension for it. So I wrote my type instrumentation described below:

package com.hello.opentelemetry; 

public class FgQueueInsertTypeInstrumentation implements TypeInstrumentation {

    public FgQueueInsertTypeInstrumentation() {
    }

    @Override
    public ElementMatcher<TypeDescription> typeMatcher() {
        return ElementMatchers.named("com.hello.FacadeMain");
    }

    @Override
    public void transform(TypeTransformer typeTransformer) {
        Junction<MethodDescription> match = ElementMatchers.namedOneOf("putData");
        typeTransformer.applyAdviceToMethod(match, this.getClass().getName() + "$FgInsertQueueAdvice");
    }

    // The FgCustomAdvice class contains the actual pieces of code (advices) that will be added to the instrumented methods.
    @SuppressWarnings("unused")
    public static class FgInsertQueueAdvice {

        @Advice.OnMethodEnter(suppress = Throwable.class)
        public static Scope onEnter(@Advice.Argument(value = 0) Object input,
                                    @Advice.Local("otelCallDepth") CallDepth callDepth,
                                    //                                    @Advice.Local("otelContext") Context context,
                                    @Advice.Local("otelSpan") Span span,
                                    @Advice.Origin("#t") Class<?> declaringClass,
                                    @Advice.Origin("#m") String methodName) {

            String spanName = declaringClass.getName() + "." + methodName;
            System.out.println("Entering method=" + spanName);
            Class<Object> class1 = (Class<Object>) input.getClass();
            System.out.println("Input parameter=" + class1.getName());

            try {
                VirtualField<Object, Integer> virtualField = VirtualField.find(class1, Integer.class);
                //                virtualField.set(class1, Integer.valueOf(50));
                System.out.println("After virtual method=" + spanName);

                throw new RuntimeException("null found");
            } catch (Throwable e) {
                e.printStackTrace();
                throw e;
            }
        }

        @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
        public static void onExit(@Advice.Thrown Throwable throwable,
                                  @Advice.Local("otelSpan") Span span,
                                  @Advice.Enter Scope scope) {
            scope.close();

            if (throwable != null) {
                span.setStatus(StatusCode.ERROR, "Exception thrown in method");
            }
            span.end();
        }
    }
}

When I run it, I received the exception on VirtualField<Object, Integer> virtualField = VirtualField.find(class1, Integer.class);

java.lang.IllegalStateException: VirtualField not found
	at io.opentelemetry.javaagent.tooling.field.RuntimeFieldBasedImplementationSupplier.find(RuntimeFieldBasedImplementationSupplier.java:30)
	at io.opentelemetry.javaagent.shaded.instrumentation.api.util.VirtualField.find(VirtualField.java:41)
	at datapump/com.hello.FacadeMain.putData(FacadeMain.java:39)
	at datapump/com.hello.datapump.FacadeMain.main(FacadeMain.java:11)
Caused by: java.lang.ClassNotFoundException: io/opentelemetry/javaagent/bootstrap/field/VirtualFieldImpl$java$lang$Integer$java$lang$Integer
	at java.base/java.lang.Class.forName0(Native Method)
	at java.base/java.lang.Class.forName(Class.java:467)
	at io.opentelemetry.javaagent.tooling.field.RuntimeFieldBasedImplementationSupplier.find(RuntimeFieldBasedImplementationSupplier.java:24)
	... 3 more

I guess I need to initialise VirtualField somehow but I don't know how to. I need advices for how to fix it.

By the way, I am using jdk:

java version "17.0.4.1" 2022-08-18 LTS
Java(TM) SE Runtime Environment (build 17.0.4.1+1-LTS-2)
Java HotSpot(TM) 64-Bit Server VM (build 17.0.4.1+1-LTS-2, mixed mode, sharing)

Thanks in advance

[laurit]

Did you build your extension following the sample extension https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/examples/extension ? VirtualField handling requires a gradle plugin to be run during the build.

[harveychong]

Thanks for your quick reply.

I have migrated gradle script from the example project to my java project for opentelemetry extension development. When I run gradle clean jar --stacktrace I received the following exception(partially):

* Exception is:
Caused by: java.lang.IllegalStateException: Failed to create ByteBuddy plugin instance
        at io.opentelemetry.javaagent.muzzle.generation.ClasspathByteBuddyPlugin$Companion.pluginFromClassPath2(ClasspathByteBuddyPlugin.kt:58)
        at io.opentelemetry.javaagent.muzzle.generation.ClasspathByteBuddyPlugin$Companion.access$pluginFromClassPath2(ClasspathByteBuddyPlugin.kt:47)
        at io.opentelemetry.javaagent.muzzle.generation.ClasspathByteBuddyPlugin.<init>(ClasspathByteBuddyPlugin.kt:29)
        at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
        at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77)
        at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at net.bytebuddy.build.Plugin$Factory$UsingReflection$Instantiator$Resolved.instantiate(Plugin.java:423)
        ... 126 more
Caused by: java.lang.ClassNotFoundException: io.opentelemetry.javaagent.tooling.muzzle.generation.MuzzleCodeGenerationPlugin
        at io.opentelemetry.javaagent.muzzle.generation.ClasspathByteBuddyPlugin$Companion.pluginFromClassPath2(ClasspathByteBuddyPlugin.kt:55)
        ... 132 more

From the stacktrace, it seems that I am still missing something as Caused by: java.lang.ClassNotFoundException: io.opentelemetry.javaagent.tooling.muzzle.generation.MuzzleCodeGenerationPlugin.

For your reference, the following is my build.gradle:

plugins {
  id "java"
  id "com.github.johnrengelman.shadow" version "8.1.1"
  id "com.diffplug.spotless" version "6.22.0"

  id "io.opentelemetry.instrumentation.muzzle-generation" version "1.32.0-alpha-SNAPSHOT"
  id "io.opentelemetry.instrumentation.muzzle-check" version "1.32.0-alpha-SNAPSHOT"
}

group = "fgp.instrumentation"
version = "0.0.1"

repositories {
    mavenCentral()
    
    maven {
    name = "sonatype"
    url = uri("https://oss.sonatype.org/content/repositories/snapshots")
  }
}

ext {
  versions = [
    opentelemetrySdk           : "1.31.0",
    opentelemetryJavaagent     : "1.32.0-SNAPSHOT",
    opentelemetryJavaagentAlpha: "1.32.0-alpha-SNAPSHOT",

    junit                      : "5.10.0"
  ]

//  deps = [
//    autoservice: dependencies.create(group: 'com.google.auto.service', name: 'auto-service', version: '1.1.1')
//  ]
}

configurations {
  otel
}

apply plugin: 'java-library'
apply plugin: 'eclipse'

spotless {
  java {
    googleJavaFormat()
    licenseHeaderFile(rootProject.file("buildscripts/spotless.license.java"), "(package|import|public)")
    target("src/**/*.java")
  }
}

dependencies {

    implementation(platform("io.opentelemetry:opentelemetry-bom:${versions.opentelemetrySdk}"))

    implementation(platform("io.opentelemetry.instrumentation:opentelemetry-instrumentation-bom:${versions.opentelemetryJavaagent}"))
    implementation(platform("io.opentelemetry.instrumentation:opentelemetry-instrumentation-bom-alpha:${versions.opentelemetryJavaagentAlpha}"))


    compileOnly('io.opentelemetry.javaagent:opentelemetry-javaagent-extension-api:1.30.0-alpha')
	compileOnly('io.opentelemetry.javaagent.instrumentation:opentelemetry-javaagent-opentelemetry-api-1.0:1.30.0-alpha')
	compileOnly('io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi:1.30.1')
	compileOnly('io.opentelemetry:opentelemetry-api:1.30.1')
	compileOnly('io.opentelemetry.instrumentation:opentelemetry-instrumentation-api:1.30.0')
	
	compileOnly('io.opentelemetry:opentelemetry-semconv:1.30.1-alpha')
	compileOnly('io.opentelemetry.javaagent:opentelemetry-javaagent-tooling:1.30.0-alpha')

    testImplementation 'org.junit.jupiter:junit-jupiter-api:5.8.2'
    testImplementation 'org.junit.jupiter:junit-jupiter-params:5.8.2'
    testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.8.2'
}

test {
    useJUnitPlatform()
}

eclipse {    
  classpath {
    // see https://docs.gradle.org/current/dsl/org.gradle.plugins.ide.eclipse.model.EclipseClasspath.html
    downloadSources = true
    downloadJavadoc = true
  }
} 

/* handle non existing source folders*/
def eclipseSourceFolders=[
    'src/main/java',
    'src/main/resources',
    'src/test/java',
    'src/test/resources'
];
tasks.eclipse.dependsOn << {
        
    for (String sourceFolder: eclipseSourceFolders){ 
        def resourceDir = new File(project.projectDir, sourceFolder)
        /* try to create folder when not existing */
        if( !resourceDir.exists() && ! resourceDir.mkdirs() ) {
            logger.info("Not able to create %1",resourceDir);
        }
    }
}

In addition, I am not sure which nominate version of OpenTelementry I should use. I don't know where the variable versions.opentelemetrySdk in build.gradle is defined.
Thanks in advance.

[harveychong]

I have added the following lines to build.gradle:

  //TODO remove when start using io.opentelemetry.instrumentation.javaagent-instrumentation plugin
  add("codegen", "io.opentelemetry.javaagent:opentelemetry-javaagent-tooling:${versions.opentelemetryJavaagentAlpha}")
  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}")

and received new exception during build

Caused by: java.lang.UnsupportedOperationException: onError
        at net.bytebuddy.build.Plugin$Engine$ErrorHandler$Failing$1.onError(Plugin.java:1186)
        at net.bytebuddy.build.Plugin$Engine$ErrorHandler$Compound.onError(Plugin.java:1438)
        at net.bytebuddy.build.Plugin$Engine$Listener$ForErrorHandler.onError(Plugin.java:2022)
        at net.bytebuddy.build.Plugin$Engine$Listener$Compound.onError(Plugin.java:2151)
        at net.bytebuddy.build.Plugin$Engine$Default$Preprocessor$Resolved.call(Plugin.java:4904)
        at net.bytebuddy.build.Plugin$Engine$Default$Preprocessor$Resolved.call(Plugin.java:4850)
        at net.bytebuddy.build.Plugin$Engine$Dispatcher$ForSerialTransformation.accept(Plugin.java:3902)
        at net.bytebuddy.build.Plugin$Engine$Default.apply(Plugin.java:4697)
        at net.bytebuddy.build.gradle.AbstractByteBuddyTask.doApply(AbstractByteBuddyTask.java:461)
        ... 123 more

[laurit]

If you want someone to look at this your best option is to produce a standalone project that reproduces the issue and upload it to a github repo. I doubt that there is anybody who can tell why this fails based solely on the available info.

[harveychong]

Thank you for your reply. I should ask a specific question. Regarding the FacadeMain application I mentioned in this discussion, do i have to use VirtualField for propagating trace information? In other words, do you know other easy method to propagate trace method rather than using the plugin? Note that we don't want to modify our application.

[laurit]

do i have to use VirtualField for propagating trace information

you don't have to use it, it is just a tool that helps writing instrumentation. You can view VirtualField as adding a field to target class that allows you to attach extra state to the object. Similar result could be achieved with a weak identity map that uses target object as key and extra state as value. Internally VirtualField implementation falls back to map if it can't add a field to the target class (happens for example when the target class is already loaded). If you so wish you can also use a similar map yourself to pass the context. Looking back at your advice code I guess it probably fails because of

VirtualField<Object, Integer> virtualField = VirtualField.find(class1, Integer.class);

we require the classes passed to VirtualField.find to be constant as we extract these values from the bytecode. If they are variables like class1 here then that extraction will fail.
Another thing to consider is that using VirtualField (or a map) is unlikely to be appropriate for the FacadeMain sample you gave. The problem is that you may pass the same Integer instance to the queue multiple times. In that case adding the same object to the queue would overwrite the context for the exiting object. It would be more appropriate to have a separate queue that tracks context. I would guess that your real application does not use integers so it probably doesn't have this issue and could use VirtualField.

do you know other easy method to propagate trace method rather than using the plugin

Please elaborate what you mean by the plugin

[harveychong]

Thank you very much for your reply.

you don't have to use it, it is just a tool that helps writing instrumentation. You can view VirtualField as adding a field to target class that allows you to attach extra state to the object.

I learned VirtualField from io.opentelemetry.javaagent.instrumentation.executors.CallableInstrumentation and I thought this is a only way to propagate context between these two threads. It seems that this way is a little bit hard for me. I am happy to try other methods.

Similar result could be achieved with a weak identity map that uses target object as key and extra state as value

Do you mean I should use TextMapPropagator ? Any example?

It would be more appropriate to have a separate queue that tracks context. I would guess that your real application does not use integers so it probably doesn't have this issue and could use VirtualField.

I would like to trace every message in the queue between these two threads. Doing so requires sharing context but I am not sure how to. This is why I went through all the classes in package io.opentelemetry.javaagent.instrumentation.executors.

Please elaborate what you mean by the plugin

I mean the extension where TypeInstrumentation and InstrumentationModule would be implemented.

[laurit]

I would like to trace every message in the queue between these two threads. Doing so requires sharing context but I am not sure how to

I'd suggest you to firstly try implementing this using manual instrumentation perhaps that would be simpler for you as you won't have to deal with bytecode modifications etc. Either add a field with context to your message or wrap it in another object that has fields for both the original message and context and see how it works out. Once you get it working try doing something similar with agent extension.