Added OpenTelemetry Support

Author: Robert Winkler

build.gradle.kts

@@ -64,7 +64,7 @@ subprojects {
     }
 
     kotlin {
-        jvmToolchain(17)
+        jvmToolchain(21)
     }
 
 }

kotlin-wot-binding-http/build.gradle.kts

@@ -14,6 +14,9 @@ dependencies {
     implementation("io.ktor:ktor-serialization-jackson")
     implementation("io.ktor:ktor-server-metrics-micrometer")
     implementation("io.ktor:ktor-server-auto-head-response")
+
+    implementation("io.opentelemetry.instrumentation:opentelemetry-ktor-3.0:2.13.1-alpha")
+
     testImplementation("io.ktor:ktor-server-test-host")
     testImplementation("ch.qos.logback:logback-classic:1.5.12")
     testImplementation("com.marcinziolo:kotlin-wiremock:2.1.1")

kotlin-wot-binding-websocket/build.gradle.kts

@@ -1,12 +1,17 @@
 dependencies {
     implementation(platform("io.ktor:ktor-bom:3.0.3"))
+
     api(project(":kotlin-wot"))
     api(project(":kotlin-wot-lmos-protocol"))
     implementation("org.slf4j:slf4j-api")
     implementation("io.ktor:ktor-server-netty")
     implementation("io.ktor:ktor-server-websockets")
     implementation("io.ktor:ktor-client-websocket:1.1.4")
     implementation("io.ktor:ktor-server-content-negotiation")
+
+    // Tracing
+    implementation("io.opentelemetry.instrumentation:opentelemetry-ktor-3.0:2.13.1-alpha")
+
     implementation("io.ktor:ktor-client-cio")
     implementation("io.ktor:ktor-client-auth")
     implementation("io.ktor:ktor-client-logging")

kotlin-wot-binding-websocket/src/main/kotlin/websocket/WebSocketProtocolClient.kt

@@ -238,7 +238,7 @@ class WebSocketProtocolClient(
         }
     }
 
-    private suspend fun requestAndReply(form: WoTForm, message: WoTMessage, timeoutMillis: Long = 10000L): Content {
+    private suspend fun requestAndReply(form: WoTForm, message: WoTMessage, timeoutMillis: Long = 100000L): Content {
         val session = getOrCreateSession(form.href)
         val deferred = CompletableDeferred<Content>()
 

kotlin-wot-binding-websocket/src/main/kotlin/websocket/WebSocketProtocolServer.kt

@@ -11,6 +11,7 @@ import ai.ancf.lmos.wot.thing.form.Form
 import ai.ancf.lmos.wot.thing.form.Operation
 import ai.ancf.lmos.wot.thing.schema.ContentListener
 import ai.ancf.lmos.wot.thing.schema.WoTExposedThing
+import ai.ancf.lmos.wot.tracing.withSpan
 import ai.anfc.lmos.wot.binding.ProtocolServer
 import ai.anfc.lmos.wot.binding.ProtocolServerException
 import com.fasterxml.jackson.databind.DeserializationFeature
@@ -31,6 +32,10 @@ import io.ktor.websocket.*
 import io.micrometer.core.instrument.binder.jvm.JvmGcMetrics
 import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics
 import io.micrometer.core.instrument.binder.system.ProcessorMetrics
+import io.opentelemetry.api.trace.Span
+import io.opentelemetry.api.trace.SpanKind
+import io.opentelemetry.instrumentation.annotations.SpanAttribute
+import io.opentelemetry.instrumentation.annotations.WithSpan
 import org.slf4j.LoggerFactory
 import java.util.*
 
@@ -154,11 +159,13 @@ class WebSocketProtocolServer(
 
 // Default server setup
 fun defaultWebSocketServer(host: String, port: Int, servient: Servient): EmbeddedServer<*, *> {
+
     return embeddedServer(Netty, port = port, host = host) {
         setupRoutingWithWebSockets(servient)
     }
 }
 
+
 fun Application.setupRoutingWithWebSockets(servient: Servient) {
     install(CallLogging)
     install(ContentNegotiation) {
@@ -196,41 +203,74 @@ fun Application.setupRoutingWithWebSockets(servient: Servient) {
             }
         }
 
+
         webSocket("/ws") {
-            val sessionId = UUID.randomUUID().toString()
-
-            for (frame in incoming) {
-                if (frame is Frame.Text) {
-                    try {
-                        // Deserialize the message to WoTMessage
-                        val message: WoTMessage = JsonMapper.instance.readValue(frame.readText())
-                        // Retrieve the thingId from the message
-                        val thingId = message.thingId
-                        val thing = servient.things[thingId]
-
-                        if (thing == null) {
-                            sendError(thingId, message.messageId, ErrorType.THING_NOT_FOUND)
-                            return@webSocket
-                        }
-
-                        // Handle the message based on its type
-                        when (message) {
-                            is ReadAllPropertiesMessage -> handleReadAllProperties(thing, message, thingId)
-                            is ReadPropertyMessage -> handleReadProperty(thing, message, thingId)
-                            is WritePropertyMessage -> handleWriteProperty(thing, message, thingId)
-                            is ObservePropertyMessage -> handleObserveProperty(thing, message, thingId, sessionId)
-                            is UnobservePropertyMessage -> handleUnobserveProperty(thing, message, thingId, sessionId)
-                            is InvokeActionMessage -> handleInvokeAction(thing, message, thingId)
-                            is SubscribeEventMessage -> handleSubscribeEvent(thing, message, thingId, sessionId)
-                            is UnsubscribeEventMessage -> handleUnsubscribeEvent(thing, message, thingId, sessionId)
-                            else -> sendError(thingId, message.messageId, ErrorType.UNSUPPORTED_MESSAGE_TYPE)
-                        }
-                    } catch (e: Exception) {
-                        val errorMessage = "Failed to read message"
-                        log.warn(errorMessage, e)
-                        close(CloseReason(CloseReason.Codes.CANNOT_ACCEPT, errorMessage))
+            val sessionId = this.call.request.headers["Sec-WebSocket-Key"] ?: UUID.randomUUID().toString()
+            handleWebSocketSession(sessionId, servient)
+        }
+    }
+}
+
+@WithSpan(kind = SpanKind.SERVER)
+suspend fun DefaultWebSocketServerSession.handleWebSocketSession(
+    @SpanAttribute("websocket.session.id") sessionId: String,
+    servient: Servient
+) {
+    for (frame in incoming) {
+        if (frame is Frame.Text) {
+            try {
+                withSpan("WebSocketProtocolServer.receiveMessage", {
+                    setSpanKind(SpanKind.SERVER)
+                }) { span ->
+                    // Deserialize the message to WoTMessage
+                    val message: WoTMessage = JsonMapper.instance.readValue(frame.readText())
+                    // Retrieve the thingId from the message
+                    val thingId = message.thingId
+                    val thing = servient.things[thingId]
+
+                    span.setAttribute("websocket.message.id", message.messageId)
+                    span.setAttribute("websocket.message.thing.id", thingId)
+                    span.setAttribute("websocket.session.id", sessionId)
+
+                    if (thing == null) {
+                        sendError(thingId, message.messageId, ErrorType.THING_NOT_FOUND)
+                        return@withSpan
+                    }
+
+                    // Handle the message based on its type
+                    when (message) {
+                        is ReadAllPropertiesMessage -> handleReadAllProperties(thing, message, thingId)
+                        is ReadPropertyMessage -> handleReadProperty(thing, message, thingId)
+                        is WritePropertyMessage -> handleWriteProperty(thing, message, thingId)
+                        is ObservePropertyMessage -> handleObserveProperty(
+                            thing,
+                            message,
+                            thingId,
+                            sessionId
+                        )
+
+                        is UnobservePropertyMessage -> handleUnobserveProperty(
+                            thing,
+                            message,
+                            thingId,
+                            sessionId
+                        )
+
+                        is InvokeActionMessage -> handleInvokeAction(thing, message, thingId)
+                        is SubscribeEventMessage -> handleSubscribeEvent(thing, message, thingId, sessionId)
+                        is UnsubscribeEventMessage -> handleUnsubscribeEvent(
+                            thing,
+                            message,
+                            thingId,
+                            sessionId
+                        )
+
+                        else -> sendError(thingId, message.messageId, ErrorType.UNSUPPORTED_MESSAGE_TYPE)
                     }
                 }
+            } catch (e: Exception) {
+                val errorMessage = "Failed to read message"
+                close(CloseReason(CloseReason.Codes.CANNOT_ACCEPT, errorMessage))
             }
         }
     }
@@ -248,6 +288,9 @@ enum class ErrorType {
     UNSUPPORTED_MESSAGE_TYPE
 }
 
+
+
+@WithSpan(kind = SpanKind.SERVER)
 suspend fun DefaultWebSocketServerSession.sendError(thingId: String, correlationId : String, errorType: ErrorType, message: String? = null) {
     val errorJson = when (errorType) {
         ErrorType.THING_NOT_FOUND -> createThingNotFound(thingId, correlationId)
@@ -262,6 +305,7 @@ suspend fun DefaultWebSocketServerSession.sendError(thingId: String, correlation
     sendSerialized(errorJson)
 }
 
+
 private suspend fun readProperty(
     thing: ExposedThing,
     propertyName: String,
@@ -275,7 +319,8 @@ private suspend fun readProperty(
     )
 }
 
-suspend fun DefaultWebSocketServerSession.handleReadAllProperties(thing: ExposedThing, message: ReadAllPropertiesMessage, thingId: String) {
+@WithSpan(kind = SpanKind.SERVER)
+suspend fun DefaultWebSocketServerSession.handleReadAllProperties(thing: ExposedThing, message: ReadAllPropertiesMessage, @SpanAttribute("thingId") thingId: String) {
 
     try {
         val propertyMap = thing.handleReadAllProperties().mapValues { entry ->
@@ -292,7 +337,8 @@ suspend fun DefaultWebSocketServerSession.handleReadAllProperties(thing: Exposed
     }
 }
 
-suspend fun DefaultWebSocketServerSession.handleReadProperty(thing: ExposedThing, message: ReadPropertyMessage, thingId: String) {
+@WithSpan(kind = SpanKind.SERVER)
+suspend fun DefaultWebSocketServerSession.handleReadProperty(thing: ExposedThing, message: ReadPropertyMessage, @SpanAttribute("thingId") thingId: String) {
     val propertyName = message.property
     val property = thing.properties[propertyName]
 
@@ -311,7 +357,8 @@ suspend fun DefaultWebSocketServerSession.handleReadProperty(thing: ExposedThing
     }
 }
 
-suspend fun DefaultWebSocketServerSession.handleWriteProperty(thing: ExposedThing, message: WritePropertyMessage, thingId: String) {
+@WithSpan(kind = SpanKind.SERVER)
+suspend fun DefaultWebSocketServerSession.handleWriteProperty(thing: ExposedThing, message: WritePropertyMessage, @SpanAttribute("thingId")thingId: String) {
     val propertyName = message.property
     val data = message.data
     val property = thing.properties[propertyName]
@@ -330,8 +377,8 @@ suspend fun DefaultWebSocketServerSession.handleWriteProperty(thing: ExposedThin
         }
     }
 }
-
-suspend fun DefaultWebSocketServerSession.handleObserveProperty(thing: ExposedThing, message: ObservePropertyMessage, thingId: String, sessionId: String) {
+@WithSpan(kind = SpanKind.SERVER)
+suspend fun DefaultWebSocketServerSession.handleObserveProperty(thing: ExposedThing, message: ObservePropertyMessage, @SpanAttribute("thingId") thingId: String, @SpanAttribute("sessionId") sessionId: String) {
     val propertyName = message.property
     val property = thing.properties[propertyName]
 
@@ -360,8 +407,8 @@ suspend fun DefaultWebSocketServerSession.handleObserveProperty(thing: ExposedTh
         }
     }
 }
-
-suspend fun DefaultWebSocketServerSession.handleUnobserveProperty(thing: ExposedThing, message: UnobservePropertyMessage, thingId: String, sessionId: String) {
+@WithSpan(kind = SpanKind.SERVER)
+suspend fun DefaultWebSocketServerSession.handleUnobserveProperty(thing: ExposedThing, message: UnobservePropertyMessage, @SpanAttribute("thingId") thingId: String, @SpanAttribute("sessionId") sessionId: String) {
     val propertyName = message.property
     val property = thing.properties[propertyName]
 
@@ -377,11 +424,16 @@ suspend fun DefaultWebSocketServerSession.handleUnobserveProperty(thing: Exposed
         }
     }
 }
+@WithSpan(kind = SpanKind.SERVER)
+suspend fun DefaultWebSocketServerSession.handleInvokeAction(thing: ExposedThing, message: InvokeActionMessage, @SpanAttribute("thingId") thingId: String) {
+
 
-suspend fun DefaultWebSocketServerSession.handleInvokeAction(thing: ExposedThing, message: InvokeActionMessage, thingId: String) {
     val actionName = message.action
     val action = thing.actions[actionName]
 
+    Span.current().setAttribute("wot.thing.id", thingId)
+    Span.current().setAttribute("wot.action.name", actionName)
+
     if (action == null) {
         sendError(thingId, message.messageId, ErrorType.ACTION_NOT_FOUND, actionName)
     } else {
@@ -401,8 +453,8 @@ suspend fun DefaultWebSocketServerSession.handleInvokeAction(thing: ExposedThing
         }
     }
 }
-
-suspend fun DefaultWebSocketServerSession.handleSubscribeEvent(thing: ExposedThing, message: SubscribeEventMessage, thingId: String, sessionId: String) {
+@WithSpan(kind = SpanKind.SERVER)
+suspend fun DefaultWebSocketServerSession.handleSubscribeEvent(thing: ExposedThing, message: SubscribeEventMessage, @SpanAttribute("thingId") thingId: String, @SpanAttribute("sessionId") sessionId: String) {
     val eventName = message.event
     val event = thing.events[eventName]
 
@@ -431,8 +483,8 @@ suspend fun DefaultWebSocketServerSession.handleSubscribeEvent(thing: ExposedThi
         }
     }
 }
-
-suspend fun DefaultWebSocketServerSession.handleUnsubscribeEvent(thing: ExposedThing, message: UnsubscribeEventMessage, thingId: String, sessionId: String) {
+@WithSpan(kind = SpanKind.SERVER)
+suspend fun DefaultWebSocketServerSession.handleUnsubscribeEvent(thing: ExposedThing, message: UnsubscribeEventMessage, @SpanAttribute("thingId") thingId: String, @SpanAttribute("sessionId") sessionId: String) {
     val eventName = message.event
     val event = thing.events[eventName]
 

kotlin-wot-integration-tests/build.gradle.kts

@@ -1,4 +1,6 @@
-import org.springframework.boot.gradle.tasks.bundling.BootJar
+
+import org.springframework.boot.gradle.tasks.run.BootRun
+import java.net.URI
 
 plugins {
     kotlin("plugin.spring") version "1.9.10"
@@ -11,21 +13,77 @@ tasks.named<Test>("test") {
 }
 
 dependencies {
+    // Replace the following with the starter dependencies of specific modules you wish to use
     api(project(":kotlin-wot-binding-http"))
     api(project(":kotlin-wot-binding-websocket"))
     api(project(":kotlin-wot-binding-mqtt"))
     api(project(":kotlin-wot-spring-boot-starter"))
     api(project(":kotlin-wot-lmos-protocol"))
-    implementation("ai.ancf.lmos:arc-azure-client:0.111.0")
-    api("ai.ancf.lmos:arc-spring-boot-starter:0.111.0")
+    implementation("org.eclipse.lmos:arc-azure-client:0.1.0-SNAPSHOT")
+
+    api("org.eclipse.lmos:arc-spring-boot-starter:0.1.0-SNAPSHOT")
+
+    //implementation("io.opentelemetry.instrumentation:opentelemetry-spring-boot-starter:2.13.1")
+    //implementation("com.azure:azure-core-metrics-opentelemetry:1.0.0-beta.27")
+    //implementation("com.azure:azure-core-tracing-opentelemetry:1.0.0-beta.55")
+    //implementation("io.opentelemetry.instrumentation:opentelemetry-instrumentation-annotations:2.13.1")
+    implementation("io.opentelemetry.instrumentation:opentelemetry-logback-appender-1.0:2.13.1-alpha")
+    //implementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure")
+    //implementation("io.opentelemetry:opentelemetry-exporter-otlp")
+
+    // AspectJ runtime for annotation-based AOP
+    //implementation("org.aspectj:aspectjrt:1.9.9")
 
-    //implementation("dev.langchain4j:langchain4j-azure-open-ai:0.35.0")
+    // AspectJ weaver for load-time weaving (LTW)
+    //implementation("org.aspectj:aspectjweaver:1.9.9")
+    //implementation("io.micrometer:micrometer-tracing-bridge-otel")
+    //implementation("io.opentelemetry:opentelemetry-exporter-otlp")
+
+    implementation("dev.langchain4j:langchain4j-azure-open-ai:1.0.0-beta1")
     //implementation("dev.langchain4j:langchain4j:0.35.0")
     testImplementation("org.springframework.boot:spring-boot-starter-test")
     testImplementation("com.hivemq:hivemq-mqtt-client:1.3.3")
     implementation("org.testcontainers:testcontainers:1.20.3")
+
+}
+
+dependencyManagement {
+    imports {
+        mavenBom("io.opentelemetry.instrumentation:opentelemetry-instrumentation-bom:2.13.1")
+    }
+}
+
+springBoot {
+    mainClass.set("ai.ancf.lmos.wot.integration.AgentApplicationKt")
+}
+
+tasks.register("downloadOtelAgent") {
+    doLast {
+        val agentUrl =
+            "https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/latest/download/opentelemetry-javaagent.jar"
+        val agentFile = file("${project.buildDir}/libs/opentelemetry-javaagent.jar")
+
+        // Ensure directory exists before downloading
+        agentFile.parentFile.mkdirs()
+
+        if (!agentFile.exists()) {
+            println("Downloading OpenTelemetry Java Agent...")
+            agentFile.writeBytes(URI(agentUrl).toURL().readBytes())
+            println("Download completed: ${agentFile.absolutePath}")
+        } else {
+            println("OpenTelemetry Java Agent already exists: ${agentFile.absolutePath}")
+        }
+    }
 }
 
-tasks.withType<BootJar> {
-    mainClass.set("integration.AgentApplication")
+tasks.named<BootRun>("bootRun") {
+    dependsOn("downloadOtelAgent")
+    jvmArgs = listOf(
+        "-javaagent:${project.buildDir}/libs/opentelemetry-javaagent.jar"
+    )
+    systemProperty("otel.java.global-autoconfigure.enabled", "true")
+    systemProperty("otel.traces.exporter", "otlp")
+    systemProperty("otel.exporter.otlp.endpoint", "http://localhost:4318")
+    systemProperty("otel.service.name", "chat-agent")
+    //systemProperty("otel.javaagent.debug", "true")
 }