Fix current service timing out (#785)

BFergerson · web-flow · commit 7a2ec2217257 · 2022-08-03T20:11:45.000+04:00
* refactor: added logging * fix: trigger indicator stopped on guide mark removed * fix: added warning for no service refactor: return null on no service instead of throwing err ref: sourceplusplus/sourceplusplus#421 * refactor: remove unused code * fix: ensure live plugins closed before opening again
diff --git a/commander b/commander
@@ -1 +1 @@
-Subproject commit b6bdfda84a101b0be7680dd153eada9d9791d982
+Subproject commit 719064832de2dc79c979a82914b01be9aa9ab162
diff --git a/marker/src/main/kotlin/spp/jetbrains/marker/source/info/EndpointDetector.kt b/marker/src/main/kotlin/spp/jetbrains/marker/source/info/EndpointDetector.kt
@@ -18,8 +18,6 @@ package spp.jetbrains.marker.source.info
 
 import io.vertx.core.Future
 import io.vertx.core.Vertx
-import io.vertx.core.eventbus.ReplyException
-import io.vertx.core.eventbus.ReplyFailure
 import io.vertx.kotlin.coroutines.await
 import org.slf4j.LoggerFactory
 import spp.jetbrains.marker.source.mark.api.key.SourceKey
@@ -105,20 +103,12 @@ abstract class EndpointDetector<T : EndpointDetector.EndpointNameDeterminer>(val
 
     private suspend fun determineEndpointId(endpointName: String, sourceMark: MethodGuideMark) {
         log.trace("Determining endpoint id")
-        try {
-            val endpoint = EndpointBridge.searchExactEndpoint(endpointName, vertx)
-            if (endpoint != null) {
-                sourceMark.putUserData(ENDPOINT_ID, endpoint.getString("id"))
-                log.trace("Detected endpoint id: ${endpoint.getString("id")}")
-            } else {
-                log.trace("Could not find endpoint id for: $endpointName")
-            }
-        } catch (ex: ReplyException) {
-            if (ex.failureType() == ReplyFailure.TIMEOUT) {
-                log.debug("Timed out looking for endpoint id for: $endpointName")
-            } else {
-                throw ex
-            }
+        val endpoint = EndpointBridge.searchExactEndpoint(endpointName, vertx)
+        if (endpoint != null) {
+            sourceMark.putUserData(ENDPOINT_ID, endpoint.getString("id"))
+            log.trace("Detected endpoint id: ${endpoint.getString("id")}")
+        } else {
+            log.trace("Could not find endpoint id for: $endpointName")
         }
     }
 
diff --git a/monitor/src/main/kotlin/spp/jetbrains/monitor/skywalking/SkywalkingMonitorService.kt b/monitor/src/main/kotlin/spp/jetbrains/monitor/skywalking/SkywalkingMonitorService.kt
@@ -59,7 +59,7 @@ interface SkywalkingMonitorService {
     suspend fun getTraces(request: GetEndpointTraces): TraceResult
     suspend fun getTraceStack(traceId: String): TraceSpanStackQueryResult
     suspend fun queryLogs(query: LogsBridge.GetEndpointLogs): AsyncResult<LogResult>
-    suspend fun getCurrentService(): Service
+    suspend fun getCurrentService(): Service?
     suspend fun getActiveServices(): List<Service>
     suspend fun getCurrentServiceInstance(): GetServiceInstancesQuery.Result?
     suspend fun getActiveServiceInstances(): List<GetServiceInstancesQuery.Result>
diff --git a/monitor/src/main/kotlin/spp/jetbrains/monitor/skywalking/bridge/EndpointBridge.kt b/monitor/src/main/kotlin/spp/jetbrains/monitor/skywalking/bridge/EndpointBridge.kt
@@ -17,8 +17,6 @@
 package spp.jetbrains.monitor.skywalking.bridge
 
 import io.vertx.core.Vertx
-import io.vertx.core.eventbus.ReplyException
-import io.vertx.core.eventbus.ReplyFailure
 import io.vertx.core.json.JsonObject
 import io.vertx.kotlin.coroutines.CoroutineVerticle
 import io.vertx.kotlin.coroutines.await
@@ -46,21 +44,10 @@ class EndpointBridge(private val skywalkingClient: SkywalkingClient) : Coroutine
 
         vertx.eventBus().localConsumer<String>(searchExactEndpointAddress) {
             launch(vertx.dispatcher()) {
-                val service = try {
-                    ServiceBridge.getCurrentService(vertx)
-                } catch (ex: ReplyException) {
-                    if (ex.failureType() == ReplyFailure.TIMEOUT) {
-                        log.debug("Timed out looking for current service")
-                        it.reply(null)
-                        return@launch
-                    } else {
-                        ex.printStackTrace()
-                        it.fail(500, ex.message)
-                        return@launch
-                    }
-                } catch (throwable: Throwable) {
-                    throwable.printStackTrace()
-                    it.fail(404, "Apache SkyWalking current service unavailable")
+                val service = ServiceBridge.getCurrentService(vertx)
+                if (service == null) {
+                    log.warn("Unable to determine current service for endpoint search")
+                    it.reply(null)
                     return@launch
                 }
 
diff --git a/monitor/src/main/kotlin/spp/jetbrains/monitor/skywalking/bridge/ServiceBridge.kt b/monitor/src/main/kotlin/spp/jetbrains/monitor/skywalking/bridge/ServiceBridge.kt
@@ -22,6 +22,7 @@ import io.vertx.kotlin.coroutines.CoroutineVerticle
 import io.vertx.kotlin.coroutines.await
 import io.vertx.kotlin.coroutines.dispatcher
 import kotlinx.coroutines.launch
+import org.slf4j.LoggerFactory
 import spp.jetbrains.monitor.skywalking.SkywalkingClient
 import spp.jetbrains.monitor.skywalking.SkywalkingClient.DurationStep
 import spp.protocol.platform.general.Service
@@ -43,63 +44,58 @@ class ServiceBridge(
     var activeServices: List<Service> = emptyList()
 
     override suspend fun start() {
-        vertx.setPeriodic(5000) { timerId ->
-            launch(vertx.dispatcher()) {
-                activeServices = skywalkingClient.run {
-                    getServices(getDuration(ZonedDateTime.now().minusMinutes(15), DurationStep.MINUTE))
-                }
-
-                if (activeServices.isNotEmpty()) {
-                    vertx.cancelTimer(timerId)
-                    vertx.eventBus().publish(activeServicesUpdatedAddress, activeServices)
-
-                    if (initServiceName != null) {
-                        currentService = activeServices.find { it.name == initServiceName }
+        if (!setCurrentService()) {
+            //periodically check for current service
+            log.info("No current service set, starting periodic check for current service")
+            vertx.setPeriodic(5000) { timerId ->
+                launch(vertx.dispatcher()) {
+                    if (setCurrentService()) {
+                        vertx.cancelTimer(timerId)
                     }
-                    if (currentService == null) {
-                        currentService = activeServices[0]
-                    }
-                    vertx.eventBus().publish(currentServiceUpdatedAddress, currentService)
                 }
             }
         }
 
         //async accessors
         vertx.eventBus().localConsumer<Boolean>(getCurrentServiceAddress) { msg ->
-            if (msg.body() && currentService == null) {
-                val consumer = currentServiceConsumer(vertx)
-                if (currentService != null) {
-                    consumer.unregister()
-                    msg.reply(currentService)
-                } else {
-                    consumer.handler {
-                        msg.reply(it.body())
-                        consumer.unregister()
-                    }
-                }
-            } else {
-                msg.reply(currentService)
+            if (currentService == null) {
+                log.warn("No current service set")
             }
+            msg.reply(currentService)
         }
         vertx.eventBus().localConsumer<Boolean>(getActiveServicesAddress) { msg ->
-            if (msg.body() && activeServices.isEmpty()) {
-                val consumer = activeServicesConsumer(vertx)
-                if (activeServices.isNotEmpty()) {
-                    consumer.unregister()
-                    msg.reply(activeServices)
-                } else {
-                    consumer.handler {
-                        msg.reply(it.body())
-                        consumer.unregister()
-                    }
-                }
-            } else {
-                msg.reply(activeServices)
+            if (activeServices.isEmpty()) {
+                log.warn("No active services set")
             }
+            msg.reply(activeServices)
         }
     }
 
+    private suspend fun setCurrentService(): Boolean {
+        activeServices = skywalkingClient.run {
+            getServices(getDuration(ZonedDateTime.now().minusMinutes(15), DurationStep.MINUTE))
+        }
+
+        if (activeServices.isNotEmpty()) {
+            vertx.eventBus().publish(activeServicesUpdatedAddress, activeServices)
+
+            if (initServiceName != null) {
+                currentService = activeServices.find { it.name == initServiceName }
+                currentService?.let { log.info("Current service set to: {}", it.name) }
+            }
+            if (currentService == null) {
+                currentService = activeServices[0]
+                log.info("Current service set to: {}", currentService!!.name)
+            }
+            vertx.eventBus().publish(currentServiceUpdatedAddress, currentService)
+            return true
+        }
+        return false
+    }
+
     companion object {
+        private val log = LoggerFactory.getLogger(ServiceBridge::class.java)
+
         private const val rootAddress = "monitor.skywalking.service"
         private const val getCurrentServiceAddress = "$rootAddress.currentService"
         private const val getActiveServicesAddress = "$rootAddress.activeServices"
@@ -114,7 +110,7 @@ class ServiceBridge(
             return vertx.eventBus().localConsumer(activeServicesUpdatedAddress)
         }
 
-        suspend fun getCurrentService(vertx: Vertx): Service {
+        suspend fun getCurrentService(vertx: Vertx): Service? {
             return vertx.eventBus()
                 .request<Service>(getCurrentServiceAddress, true).await().body()
         }
diff --git a/monitor/src/main/kotlin/spp/jetbrains/monitor/skywalking/impl/SkywalkingMonitorServiceImpl.kt b/monitor/src/main/kotlin/spp/jetbrains/monitor/skywalking/impl/SkywalkingMonitorServiceImpl.kt
@@ -53,7 +53,8 @@ class SkywalkingMonitorServiceImpl(
     }
 
     override suspend fun searchExactEndpoint(keyword: String, cache: Boolean): JsonObject? {
-        val endpoints = skywalkingClient.searchEndpoint(keyword, getCurrentService().id, 1, cache)
+        val service = getCurrentService() ?: return null
+        val endpoints = skywalkingClient.searchEndpoint(keyword, service.id, 1, cache)
         return endpoints.map { it as JsonObject }.find { it.getString("name") == keyword }
     }
 
@@ -77,7 +78,7 @@ class SkywalkingMonitorServiceImpl(
         return EndpointTracesBridge.getTraceStack(traceId, skywalkingClient.vertx)
     }
 
-    override suspend fun getCurrentService(): Service {
+    override suspend fun getCurrentService(): Service? {
         return ServiceBridge.getCurrentService(skywalkingClient.vertx)
     }
 
diff --git a/plugin/src/main/kotlin/spp/jetbrains/sourcemarker/SourceMarkerPlugin.kt b/plugin/src/main/kotlin/spp/jetbrains/sourcemarker/SourceMarkerPlugin.kt
@@ -133,6 +133,7 @@ object SourceMarkerPlugin {
     ) {
         log.info("Initializing SourceMarkerPlugin on project: {}", project)
         restartIfNecessary()
+        LivePluginProjectLoader.projectClosing(project)
         LivePluginProjectLoader.projectOpened(project)
 
         val config = configInput ?: getConfig(project)
diff --git a/plugin/src/main/kotlin/spp/jetbrains/sourcemarker/mark/SourceMarkSearch.kt b/plugin/src/main/kotlin/spp/jetbrains/sourcemarker/mark/SourceMarkSearch.kt
@@ -20,14 +20,10 @@ import com.intellij.openapi.editor.Editor
 import com.intellij.openapi.util.TextRange
 import spp.jetbrains.marker.SourceMarker
 import spp.jetbrains.marker.source.SourceFileMarker
-import spp.jetbrains.marker.source.mark.api.MethodSourceMark
 import spp.jetbrains.marker.source.mark.api.SourceMark
 import spp.jetbrains.marker.source.mark.guide.ClassGuideMark
-import spp.jetbrains.marker.source.mark.guide.ExpressionGuideMark
 import spp.jetbrains.marker.source.mark.guide.GuideMark
 import spp.jetbrains.marker.source.mark.guide.MethodGuideMark
-import spp.protocol.artifact.ArtifactQualifiedName
-import spp.protocol.artifact.ArtifactType
 
 /**
  * todo: description.
@@ -61,12 +57,6 @@ object SourceMarkSearch {
         return sourceMark ?: classSourceMark
     }
 
-    fun findByEndpointName(endpointName: String): GuideMark? {
-        return SourceMarker.getSourceMarks().filterIsInstance<GuideMark>().firstOrNull {
-            it.getUserData(SourceMarkKeys.ENDPOINT_DETECTOR)?.getEndpointName(it) == endpointName
-        }
-    }
-
     fun findByInstrumentId(instrumentId: String): SourceMark? {
         return SourceMarker.getSourceMarks().firstOrNull {
             it.getUserData(SourceMarkKeys.INSTRUMENT_ID) == instrumentId
@@ -78,71 +68,4 @@ object SourceMarkSearch {
             it.getUserData(SourceMarkKeys.VIEW_SUBSCRIPTION_ID) == subscriptionId
         }
     }
-
-    suspend fun findSourceMark(artifact: ArtifactQualifiedName): GuideMark? {
-        return when (artifact.type) {
-            ArtifactType.ENDPOINT -> findEndpointSourceMark(artifact)
-            ArtifactType.STATEMENT -> findExpressionSourceMark(artifact)
-            ArtifactType.EXPRESSION -> findExpressionSourceMark(artifact)
-            else -> TODO("impl")
-        }
-    }
-
-    fun findSourceMarks(artifact: ArtifactQualifiedName): List<SourceMark> {
-        return SourceMarker.getSourceMarks().filter { it.artifactQualifiedName == artifact }
-    }
-
-    suspend fun findSourceMark(logPattern: String): MethodSourceMark? {
-        return SourceMarker.getSourceMarks()
-            .filterIsInstance<MethodSourceMark>()
-            .firstOrNull {
-                it.getUserData(SourceMarkKeys.LOGGER_DETECTOR)!!.getOrFindLoggerStatements(it)
-                    .map { it.logPattern }.contains(logPattern)
-            }
-    }
-
-    suspend fun findInheritedSourceMarks(logPattern: String): List<SourceMark> {
-        val rootMark = findSourceMark(logPattern) ?: return emptyList()
-        return findInheritedSourceMarks(rootMark)
-    }
-
-    fun findInheritedSourceMarks(rootMark: SourceMark): List<SourceMark> {
-        return if (rootMark.isExpressionMark) {
-            val methodMark = SourceMarker.getSourceMark(
-                rootMark.artifactQualifiedName.copy(
-                    identifier = rootMark.artifactQualifiedName.identifier.substringBefore("#"),
-                    type = ArtifactType.METHOD
-                ),
-                SourceMark.Type.GUTTER
-            )
-            //todo: proper class crawl
-            listOfNotNull(rootMark, methodMark) + rootMark.sourceFileMarker.getClassSourceMarks()
-        } else if (rootMark.isMethodMark) {
-            //todo: proper class crawl
-            listOf(rootMark) + rootMark.sourceFileMarker.getClassSourceMarks()
-        } else {
-            listOf(rootMark)
-        }
-    }
-
-    private suspend fun findEndpointSourceMark(artifact: ArtifactQualifiedName): MethodGuideMark? {
-        val operationName = artifact.identifier
-        return SourceMarker.getSourceMarks()
-            .filterIsInstance<MethodGuideMark>()
-            .firstOrNull {
-                it.getUserData(SourceMarkKeys.ENDPOINT_DETECTOR)!!.getOrFindEndpointName(it) == operationName
-            }
-    }
-
-    private fun findExpressionSourceMark(artifact: ArtifactQualifiedName): ExpressionGuideMark? {
-        if (artifact.type == ArtifactType.EXPRESSION) {
-            return SourceMarker.getSourceMarks().filterIsInstance<ExpressionGuideMark>()
-                .find { it.artifactQualifiedName == artifact }
-        }
-
-        val qualifiedClassName = artifact.identifier.substring(0, artifact.identifier.lastIndexOf("."))
-        val fileMarker = SourceMarker.getSourceFileMarker(qualifiedClassName)
-        return fileMarker?.getSourceMarks()?.filterIsInstance<ExpressionGuideMark>()
-            ?.find { it.lineNumber == artifact.lineNumber!! }
-    }
 }
diff --git a/plugin/src/main/resources/logback.xml b/plugin/src/main/resources/logback.xml
@@ -17,6 +17,12 @@
     <logger name="spp" level="DEBUG">
         <appender-ref ref="STDOUT"/>
     </logger>
+    <logger name="liveplugin" level="INFO">
+        <appender-ref ref="debug-console"/>
+    </logger>
+    <logger name="liveplugin" level="DEBUG">
+        <appender-ref ref="STDOUT"/>
+    </logger>
 
     <logger name="io.vertx" level="DEBUG">
         <appender-ref ref="STDOUT"/>

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,8 @@ class SkywalkingMonitorServiceImpl(`
`53`	`53`	`}`
`54`	`54`
`55`	`55`	`override suspend fun searchExactEndpoint(keyword: String, cache: Boolean): JsonObject? {`
`56`		`- val endpoints = skywalkingClient.searchEndpoint(keyword, getCurrentService().id, 1, cache)`
	`56`	`+ val service = getCurrentService() ?: return null`
	`57`	`+ val endpoints = skywalkingClient.searchEndpoint(keyword, service.id, 1, cache)`
`57`	`58`	`return endpoints.map { it as JsonObject }.find { it.getString("name") == keyword }`
`58`	`59`	`}`
`59`	`60`
`@@ -77,7 +78,7 @@ class SkywalkingMonitorServiceImpl(`
`77`	`78`	`return EndpointTracesBridge.getTraceStack(traceId, skywalkingClient.vertx)`
`78`	`79`	`}`
`79`	`80`
`80`		`- override suspend fun getCurrentService(): Service {`
	`81`	`+ override suspend fun getCurrentService(): Service? {`
`81`	`82`	`return ServiceBridge.getCurrentService(skywalkingClient.vertx)`
`82`	`83`	`}`
`83`	`84`