Skip to content

Commit 469f1d9

Browse files
BFergersonBFergerson
authored
Merge pull request #38 from sourceplusplus/dev
v0.4.0
2 parents 861507c + 5ada2a4 commit 469f1d9

26 files changed

+110
-207
lines changed

‎control/src/main/kotlin/spp/probe/SourceProbe.kt

+14-26
Original file line numberDiff line numberDiff line change
@@ -35,8 +35,8 @@ import spp.probe.control.LiveInstrumentRemote
3535
import spp.probe.util.NopInternalLogger
3636
import spp.probe.util.NopLogDelegateFactory
3737
import spp.protocol.platform.PlatformAddress
38-
import spp.protocol.probe.ProbeAddress
39-
import spp.protocol.probe.status.ProbeConnection
38+
import spp.protocol.platform.ProbeAddress
39+
import spp.protocol.platform.status.InstanceConnection
4040
import java.io.File
4141
import java.io.FileOutputStream
4242
import java.io.IOException
@@ -52,7 +52,7 @@ object SourceProbe {
5252

5353
private val BUILD = ResourceBundle.getBundle("build")
5454
private var PROBE_DIRECTORY = File(
55-
if (System.getProperty("os.name").lowercase(Locale.getDefault()).startsWith("mac"))
55+
if (System.getProperty("os.name").lowercase().startsWith("mac"))
5656
"/tmp" else System.getProperty("java.io.tmpdir"), "spp-probe"
5757
)
5858
var instrumentation: Instrumentation? = null
@@ -209,19 +209,19 @@ object SourceProbe {
209209
if ("message" == frame.getString("type")) {
210210
if (frame.getString("replyAddress") != null) {
211211
vertx!!.eventBus().request<Any?>(
212-
"local." + frame.getString("address"),
212+
frame.getString("address"),
213213
frame.getJsonObject("body")
214214
).onComplete {
215215
if (it.succeeded()) {
216216
FrameHelper.sendFrame(
217-
BridgeEventType.SEND.name.lowercase(Locale.getDefault()),
217+
BridgeEventType.SEND.name.lowercase(),
218218
frame.getString("replyAddress"),
219219
JsonObject.mapFrom(it.result().body()),
220220
socket.result()
221221
)
222222
} else {
223223
FrameHelper.sendFrame(
224-
BridgeEventType.SEND.name.lowercase(Locale.getDefault()),
224+
BridgeEventType.SEND.name.lowercase(),
225225
frame.getString("replyAddress"),
226226
JsonObject.mapFrom(it.cause()),
227227
socket.result()
@@ -230,7 +230,7 @@ object SourceProbe {
230230
}
231231
} else {
232232
vertx!!.eventBus().publish(
233-
"local." + frame.getString("address"),
233+
frame.getString("address"),
234234
frame.getValue("body")
235235
)
236236
}
@@ -259,40 +259,28 @@ object SourceProbe {
259259

260260
//send probe connected status
261261
val replyAddress = UUID.randomUUID().toString()
262-
val pc = ProbeConnection(PROBE_ID, System.currentTimeMillis(), meta)
263-
val consumer = vertx!!.eventBus().localConsumer<Boolean>("local.$replyAddress")
262+
val pc = InstanceConnection(PROBE_ID, System.currentTimeMillis(), meta)
263+
val consumer = vertx!!.eventBus().localConsumer<Boolean>(replyAddress)
264264
consumer.handler {
265265
if (ProbeConfiguration.isNotQuite) println("Received probe connection confirmation")
266266

267267
//register remotes
268268
FrameHelper.sendFrame(
269-
BridgeEventType.REGISTER.name.lowercase(Locale.getDefault()),
270-
ProbeAddress.LIVE_BREAKPOINT_REMOTE.address + ":" + PROBE_ID,
269+
BridgeEventType.REGISTER.name.lowercase(),
270+
ProbeAddress.LIVE_INSTRUMENT_REMOTE,
271271
JsonObject(),
272272
tcpSocket
273273
)
274274
FrameHelper.sendFrame(
275-
BridgeEventType.REGISTER.name.lowercase(Locale.getDefault()),
276-
ProbeAddress.LIVE_LOG_REMOTE.address + ":" + PROBE_ID,
277-
JsonObject(),
278-
tcpSocket
279-
)
280-
FrameHelper.sendFrame(
281-
BridgeEventType.REGISTER.name.lowercase(Locale.getDefault()),
282-
ProbeAddress.LIVE_METER_REMOTE.address + ":" + PROBE_ID,
283-
JsonObject(),
284-
tcpSocket
285-
)
286-
FrameHelper.sendFrame(
287-
BridgeEventType.REGISTER.name.lowercase(Locale.getDefault()),
288-
ProbeAddress.LIVE_SPAN_REMOTE.address + ":" + PROBE_ID,
275+
BridgeEventType.REGISTER.name.lowercase(),
276+
ProbeAddress.LIVE_INSTRUMENT_REMOTE + ":" + PROBE_ID,
289277
JsonObject(),
290278
tcpSocket
291279
)
292280
consumer.unregister()
293281
}
294282
FrameHelper.sendFrame(
295-
BridgeEventType.SEND.name.lowercase(Locale.getDefault()), PlatformAddress.PROBE_CONNECTED.address,
283+
BridgeEventType.SEND.name.lowercase(), PlatformAddress.PROBE_CONNECTED,
296284
replyAddress, JsonObject(), true, JsonObject.mapFrom(pc), socket.result()
297285
)
298286
}

‎control/src/main/kotlin/spp/probe/control/LiveInstrumentRemote.kt

+31-54
Original file line numberDiff line numberDiff line change
@@ -19,29 +19,24 @@ package spp.probe.control
1919

2020
import io.vertx.core.AbstractVerticle
2121
import io.vertx.core.eventbus.Message
22-
import io.vertx.core.json.Json
2322
import io.vertx.core.json.JsonObject
2423
import io.vertx.ext.bridge.BridgeEventType
2524
import io.vertx.ext.eventbus.bridge.tcp.impl.protocol.FrameHelper
2625
import org.apache.skywalking.apm.agent.core.context.util.ThrowableTransformer
2726
import org.apache.skywalking.apm.agent.core.plugin.WitnessFinder
2827
import spp.probe.ProbeConfiguration
2928
import spp.probe.SourceProbe
30-
import spp.protocol.instrument.LiveInstrument
31-
import spp.protocol.instrument.breakpoint.LiveBreakpoint
32-
import spp.protocol.instrument.log.LiveLog
33-
import spp.protocol.instrument.meter.LiveMeter
34-
import spp.protocol.instrument.span.LiveSpan
35-
import spp.protocol.platform.PlatformAddress
36-
import spp.protocol.probe.ProbeAddress
37-
import spp.protocol.probe.command.LiveInstrumentCommand
38-
import spp.protocol.probe.command.LiveInstrumentCommand.CommandType
29+
import spp.protocol.ProtocolMarshaller
30+
import spp.protocol.instrument.*
31+
import spp.protocol.instrument.command.CommandType
32+
import spp.protocol.instrument.command.LiveInstrumentCommand
33+
import spp.protocol.platform.ProbeAddress
34+
import spp.protocol.platform.ProcessorAddress
3935
import java.lang.instrument.Instrumentation
4036
import java.lang.reflect.InvocationTargetException
4137
import java.lang.reflect.Method
4238
import java.util.*
4339
import java.util.function.BiConsumer
44-
import kotlin.reflect.KClass
4540

4641
class LiveInstrumentRemote : AbstractVerticle() {
4742

@@ -99,84 +94,66 @@ class LiveInstrumentRemote : AbstractVerticle() {
9994
e.printStackTrace()
10095
throw RuntimeException(e)
10196
}
102-
vertx.eventBus()
103-
.localConsumer<JsonObject>("local." + ProbeAddress.LIVE_BREAKPOINT_REMOTE.address + ":" + SourceProbe.PROBE_ID)
104-
.handler { handleInstrumentationRequest(LiveBreakpoint::class, it) }
105-
vertx.eventBus()
106-
.localConsumer<JsonObject>("local." + ProbeAddress.LIVE_LOG_REMOTE.address + ":" + SourceProbe.PROBE_ID)
107-
.handler { handleInstrumentationRequest(LiveLog::class, it) }
108-
vertx.eventBus()
109-
.localConsumer<JsonObject>("local." + ProbeAddress.LIVE_METER_REMOTE.address + ":" + SourceProbe.PROBE_ID)
110-
.handler { handleInstrumentationRequest(LiveMeter::class, it) }
111-
vertx.eventBus()
112-
.localConsumer<JsonObject>("local." + ProbeAddress.LIVE_SPAN_REMOTE.address + ":" + SourceProbe.PROBE_ID)
113-
.handler { handleInstrumentationRequest(LiveSpan::class, it) }
97+
98+
vertx.eventBus() //global instrument remote
99+
.localConsumer<JsonObject>(ProbeAddress.LIVE_INSTRUMENT_REMOTE)
100+
.handler { handleInstrumentationRequest(it) }
101+
vertx.eventBus() //probe specific instrument remote
102+
.localConsumer<JsonObject>(ProbeAddress.LIVE_INSTRUMENT_REMOTE + ":" + SourceProbe.PROBE_ID)
103+
.handler { handleInstrumentationRequest(it) }
114104
}
115105

116-
private fun handleInstrumentationRequest(clazz: KClass<out LiveInstrument>, it: Message<JsonObject>) {
106+
private fun handleInstrumentationRequest(it: Message<JsonObject>) {
117107
try {
118-
val command = Json.decodeValue(it.body().toString(), LiveInstrumentCommand::class.java)
108+
val command = ProtocolMarshaller.deserializeLiveInstrumentCommand(it.body())
119109
when (command.commandType) {
120-
CommandType.ADD_LIVE_INSTRUMENT -> addInstrument(clazz, command)
110+
CommandType.ADD_LIVE_INSTRUMENT -> addInstrument(command)
121111
CommandType.REMOVE_LIVE_INSTRUMENT -> removeInstrument(command)
122112
}
123113
} catch (ex: InvocationTargetException) {
124114
if (ex.cause != null) {
125-
publishCommandError(it, ex.cause!!, clazz)
115+
publishCommandError(it, ex.cause!!)
126116
} else {
127-
publishCommandError(it, ex.targetException, clazz)
117+
publishCommandError(it, ex.targetException)
128118
}
129119
} catch (ex: Throwable) {
130-
publishCommandError(it, ex, clazz)
120+
publishCommandError(it, ex)
131121
}
132122
}
133123

134-
private fun publishCommandError(it: Message<JsonObject>, ex: Throwable, clazz: KClass<out LiveInstrument>) {
124+
private fun publishCommandError(it: Message<JsonObject>, ex: Throwable) {
135125
val map: MutableMap<String, Any> = HashMap()
136126
map["command"] = it.body().toString()
137127
map["occurredAt"] = System.currentTimeMillis()
138128
map["cause"] = ThrowableTransformer.INSTANCE.convert2String(ex, 4000)
139129

140-
val address = when (clazz) {
141-
LiveBreakpoint::class -> PlatformAddress.LIVE_BREAKPOINT_REMOVED.address
142-
LiveLog::class -> PlatformAddress.LIVE_LOG_REMOVED.address
143-
LiveMeter::class -> PlatformAddress.LIVE_METER_REMOVED.address
144-
LiveSpan::class -> PlatformAddress.LIVE_SPAN_REMOVED.address
145-
else -> throw IllegalArgumentException("Unknown instrument: $clazz")
146-
}
147130
FrameHelper.sendFrame(
148-
BridgeEventType.PUBLISH.name.lowercase(), address, JsonObject.mapFrom(map), SourceProbe.tcpSocket
131+
BridgeEventType.PUBLISH.name.lowercase(), ProcessorAddress.LIVE_INSTRUMENT_REMOVED,
132+
JsonObject.mapFrom(map), SourceProbe.tcpSocket
149133
)
150134
}
151135

152-
private fun addInstrument(clazz: KClass<out LiveInstrument>, command: LiveInstrumentCommand) {
136+
private fun addInstrument(command: LiveInstrumentCommand) {
153137
if (ProbeConfiguration.isNotQuite) println("Adding instrument: $command")
154-
val instrumentData = command.context.liveInstruments[0]
155-
applyInstrument!!.invoke(null, Json.decodeValue(instrumentData, clazz.java))
138+
applyInstrument!!.invoke(null, command.instruments.first()) //todo: check for multiple
156139
}
157140

158141
private fun removeInstrument(command: LiveInstrumentCommand) {
159-
for (breakpointData in command.context.liveInstruments) {
160-
val breakpointObject = JsonObject(breakpointData)
161-
val breakpointId = breakpointObject.getString("id")
162-
val location = breakpointObject.getJsonObject("location")
163-
val source = location.getString("source")
164-
val line = location.getInteger("line")
165-
removeInstrument!!.invoke(null, source, line, breakpointId)
142+
for (breakpoint in command.instruments) {
143+
val breakpointId = breakpoint.id
144+
val location = breakpoint.location
145+
removeInstrument!!.invoke(null, location.source, location.line, breakpointId)
166146
}
167-
for (locationData in command.context.locations) {
168-
val location = JsonObject(locationData)
169-
val source = location.getString("source")
170-
val line = location.getInteger("line")
171-
removeInstrument!!.invoke(null, source, line, null)
147+
for (location in command.locations) {
148+
removeInstrument!!.invoke(null, location.source, location.line, null)
172149
}
173150
}
174151

175152
companion object {
176153
private val EVENT_CONSUMER = BiConsumer(fun(address: String?, json: String?) {
177154
if (ProbeConfiguration.isNotQuite) println("Publishing event: $address, $json")
178155
FrameHelper.sendFrame(
179-
BridgeEventType.PUBLISH.name.lowercase(Locale.getDefault()),
156+
BridgeEventType.PUBLISH.name.lowercase(),
180157
address,
181158
JsonObject(json),
182159
SourceProbe.tcpSocket

‎control/src/test/kotlin/integration/LiveBreakpointTest.kt

+13-15
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
*/
1818
package integration
1919

20-
import io.vertx.core.Promise
2120
import io.vertx.core.json.Json
2221
import io.vertx.core.json.JsonObject
2322
import io.vertx.junit5.VertxTestContext
@@ -26,21 +25,20 @@ import kotlinx.coroutines.runBlocking
2625
import org.junit.jupiter.api.Assertions.assertEquals
2726
import org.junit.jupiter.api.Assertions.assertNotNull
2827
import org.junit.jupiter.api.Test
29-
import spp.protocol.SourceMarkerServices.Provide
30-
import spp.protocol.instrument.LiveInstrument
31-
import spp.protocol.instrument.LiveInstrumentEvent
32-
import spp.protocol.instrument.LiveInstrumentEventType
28+
import spp.protocol.SourceServices.Provide.toLiveInstrumentSubscriberAddress
29+
import spp.protocol.instrument.LiveBreakpoint
3330
import spp.protocol.instrument.LiveSourceLocation
34-
import spp.protocol.instrument.breakpoint.LiveBreakpoint
35-
import spp.protocol.instrument.breakpoint.event.LiveBreakpointHit
31+
import spp.protocol.instrument.event.LiveBreakpointHit
32+
import spp.protocol.instrument.event.LiveInstrumentEvent
33+
import spp.protocol.instrument.event.LiveInstrumentEventType
3634
import java.util.concurrent.TimeUnit
3735

3836
class LiveBreakpointTest : ProbeIntegrationTest() {
3937

4038
@Test
4139
fun testPrimitives() = runBlocking {
4240
val testContext = VertxTestContext()
43-
val consumer = vertx.eventBus().localConsumer<JsonObject>("local." + Provide.LIVE_INSTRUMENT_SUBSCRIBER)
41+
val consumer = vertx.eventBus().localConsumer<JsonObject>(toLiveInstrumentSubscriberAddress("system"))
4442
consumer.handler {
4543
testContext.verify {
4644
val event = Json.decodeValue(it.body().toString(), LiveInstrumentEvent::class.java)
@@ -56,14 +54,14 @@ class LiveBreakpointTest : ProbeIntegrationTest() {
5654
}
5755
}
5856

59-
val promise = Promise.promise<LiveInstrument>()
60-
instrumentService.addLiveInstrument(
61-
LiveBreakpoint(
62-
location = LiveSourceLocation("VariableTests", 35),
63-
applyImmediately = true
64-
), promise
57+
assertNotNull(
58+
instrumentService.addLiveInstrument(
59+
LiveBreakpoint(
60+
location = LiveSourceLocation("VariableTests", 35),
61+
applyImmediately = true
62+
)
63+
).await()
6564
)
66-
assertNotNull(promise.future().await())
6765

6866
callVariableTests()
6967
if (testContext.awaitCompletion(60, TimeUnit.SECONDS)) {

‎control/src/test/kotlin/integration/LiveLogTest.kt

+16-18
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
*/
1818
package integration
1919

20-
import io.vertx.core.Promise
2120
import io.vertx.core.json.Json
2221
import io.vertx.core.json.JsonObject
2322
import io.vertx.junit5.VertxTestContext
@@ -26,45 +25,44 @@ import kotlinx.coroutines.runBlocking
2625
import org.junit.jupiter.api.Assertions.assertEquals
2726
import org.junit.jupiter.api.Assertions.assertNotNull
2827
import org.junit.jupiter.api.Test
29-
import spp.protocol.SourceMarkerServices.Provide
30-
import spp.protocol.instrument.LiveInstrument
31-
import spp.protocol.instrument.LiveInstrumentEvent
32-
import spp.protocol.instrument.LiveInstrumentEventType
28+
import spp.protocol.SourceServices.Provide.toLiveInstrumentSubscriberAddress
29+
import spp.protocol.instrument.LiveLog
3330
import spp.protocol.instrument.LiveSourceLocation
34-
import spp.protocol.instrument.log.LiveLog
35-
import spp.protocol.instrument.log.event.LiveLogHit
31+
import spp.protocol.instrument.event.LiveInstrumentEvent
32+
import spp.protocol.instrument.event.LiveInstrumentEventType
33+
import spp.protocol.instrument.event.LiveLogHit
3634
import java.util.concurrent.TimeUnit
3735

3836
class LiveLogTest : ProbeIntegrationTest() {
3937

4038
@Test
4139
fun testPrimitives() = runBlocking {
4240
val testContext = VertxTestContext()
43-
val consumer = vertx.eventBus().localConsumer<JsonObject>("local." + Provide.LIVE_INSTRUMENT_SUBSCRIBER)
41+
val consumer = vertx.eventBus().localConsumer<JsonObject>(toLiveInstrumentSubscriberAddress("system"))
4442
consumer.handler {
4543
testContext.verify {
4644
val event = Json.decodeValue(it.body().toString(), LiveInstrumentEvent::class.java)
4745
log.trace("Received event: $event")
4846

4947
if (event.eventType == LiveInstrumentEventType.LOG_HIT) {
5048
val item = Json.decodeValue(event.data, LiveLogHit::class.java)
51-
assertEquals("1 a a", item.logResult.logs.first().getFormattedMessage())
49+
assertEquals("1 a a", item.logResult.logs.first().toFormattedMessage())
5250
}
5351
consumer.unregister()
5452
testContext.completeNow()
5553
}
5654
}
5755

58-
val promise = Promise.promise<LiveInstrument>()
59-
instrumentService.addLiveInstrument(
60-
LiveLog(
61-
logFormat = "{} {} {}",
62-
logArguments = listOf("a", "b", "c"),
63-
location = LiveSourceLocation("VariableTests", 35),
64-
applyImmediately = true
65-
), promise
56+
assertNotNull(
57+
instrumentService.addLiveInstrument(
58+
LiveLog(
59+
logFormat = "{} {} {}",
60+
logArguments = listOf("a", "b", "c"),
61+
location = LiveSourceLocation("VariableTests", 35),
62+
applyImmediately = true
63+
)
64+
).await()
6665
)
67-
assertNotNull(promise.future().await())
6866

6967
callVariableTests()
7068
if (testContext.awaitCompletion(60, TimeUnit.SECONDS)) {

0 commit comments

Comments
 (0)