1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
package akka.kamon.instrumentation
import java.util.concurrent.locks.ReentrantLock
import akka.actor._
import akka.dispatch.Envelope
import akka.dispatch.sysmsg.SystemMessage
import akka.routing.RoutedActorCell
import kamon.trace.Tracer
import org.aspectj.lang.ProceedingJoinPoint
import org.aspectj.lang.annotation._
import scala.collection.immutable
@Aspect
class ActorCellInstrumentation {
def actorInstrumentation(cell: Cell): ActorMonitor =
cell.asInstanceOf[ActorInstrumentationAware].actorInstrumentation
@Pointcut("execution(akka.actor.ActorCell.new(..)) && this(cell) && args(system, ref, *, *, parent)")
def actorCellCreation(cell: Cell, system: ActorSystem, ref: ActorRef, parent: InternalActorRef): Unit = {}
@Pointcut("execution(akka.actor.UnstartedCell.new(..)) && this(cell) && args(system, ref, *, parent)")
def repointableActorRefCreation(cell: Cell, system: ActorSystem, ref: ActorRef, parent: InternalActorRef): Unit = {}
@After("actorCellCreation(cell, system, ref, parent)")
def afterCreation(cell: Cell, system: ActorSystem, ref: ActorRef, parent: ActorRef): Unit = {
cell.asInstanceOf[ActorInstrumentationAware].setActorInstrumentation(
ActorMonitor.createActorMonitor(cell, system, ref, parent))
}
@After("repointableActorRefCreation(cell, system, ref, parent)")
def afterRepointableActorRefCreation(cell: Cell, system: ActorSystem, ref: ActorRef, parent: ActorRef): Unit = {
cell.asInstanceOf[ActorInstrumentationAware].setActorInstrumentation(
ActorMonitor.createActorMonitor(cell, system, ref, parent))
}
@Pointcut("execution(* akka.actor.ActorCell.invoke(*)) && this(cell) && args(envelope)")
def invokingActorBehaviourAtActorCell(cell: ActorCell, envelope: Envelope) = {}
@Around("invokingActorBehaviourAtActorCell(cell, envelope)")
def aroundBehaviourInvoke(pjp: ProceedingJoinPoint, cell: ActorCell, envelope: Envelope): Any = {
actorInstrumentation(cell).processMessage(pjp, envelope.asInstanceOf[InstrumentedEnvelope].envelopeContext())
}
/**
*
*
*/
@Pointcut("execution(* akka.actor.ActorCell.sendMessage(*)) && this(cell) && args(envelope)")
def sendMessageInActorCell(cell: Cell, envelope: Envelope): Unit = {}
@Pointcut("execution(* akka.actor.UnstartedCell.sendMessage(*)) && this(cell) && args(envelope)")
def sendMessageInUnstartedActorCell(cell: Cell, envelope: Envelope): Unit = {}
@Before("sendMessageInActorCell(cell, envelope)")
def afterSendMessageInActorCell(cell: Cell, envelope: Envelope): Unit = {
envelope.asInstanceOf[InstrumentedEnvelope].setEnvelopeContext(
actorInstrumentation(cell).captureEnvelopeContext())
}
@Before("sendMessageInUnstartedActorCell(cell, envelope)")
def afterSendMessageInUnstartedActorCell(cell: Cell, envelope: Envelope): Unit = {
envelope.asInstanceOf[InstrumentedEnvelope].setEnvelopeContext(
actorInstrumentation(cell).captureEnvelopeContext())
}
@Pointcut("execution(* akka.actor.UnstartedCell.replaceWith(*)) && this(unStartedCell) && args(cell)")
def replaceWithInRepointableActorRef(unStartedCell: UnstartedCell, cell: Cell): Unit = {}
@Around("replaceWithInRepointableActorRef(unStartedCell, cell)")
def aroundReplaceWithInRepointableActorRef(pjp: ProceedingJoinPoint, unStartedCell: UnstartedCell, cell: Cell): Unit = {
// TODO: Find a way to do this without resorting to reflection and, even better, without copy/pasting the Akka Code!
val unstartedCellClass = classOf[UnstartedCell]
val queueField = unstartedCellClass.getDeclaredField("akka$actor$UnstartedCell$$queue")
queueField.setAccessible(true)
val lockField = unstartedCellClass.getDeclaredField("lock")
lockField.setAccessible(true)
val queue = queueField.get(unStartedCell).asInstanceOf[java.util.LinkedList[_]]
val lock = lockField.get(unStartedCell).asInstanceOf[ReentrantLock]
def locked[T](body: ⇒ T): T = {
lock.lock()
try body finally lock.unlock()
}
locked {
try {
while (!queue.isEmpty) {
queue.poll() match {
case s: SystemMessage ⇒ cell.sendSystemMessage(s) // TODO: ============= CHECK SYSTEM MESSAGESSSSS =========
case e: Envelope with InstrumentedEnvelope ⇒
Tracer.withContext(e.envelopeContext().context) {
cell.sendMessage(e)
}
}
}
} finally {
unStartedCell.self.swapCell(cell)
}
}
}
/**
*
*/
@Pointcut("execution(* akka.actor.ActorCell.stop()) && this(cell)")
def actorStop(cell: ActorCell): Unit = {}
@After("actorStop(cell)")
def afterStop(cell: ActorCell): Unit = {
actorInstrumentation(cell).cleanup()
// The Stop can't be captured from the RoutedActorCell so we need to put this piece of cleanup here.
if (cell.isInstanceOf[RoutedActorCell]) {
cell.asInstanceOf[RouterInstrumentationAware].routerInstrumentation.cleanup()
}
}
@Pointcut("execution(* akka.actor.ActorCell.handleInvokeFailure(..)) && this(cell) && args(childrenNotToSuspend, failure)")
def actorInvokeFailure(cell: ActorCell, childrenNotToSuspend: immutable.Iterable[ActorRef], failure: Throwable): Unit = {}
@Before("actorInvokeFailure(cell, childrenNotToSuspend, failure)")
def beforeInvokeFailure(cell: ActorCell, childrenNotToSuspend: immutable.Iterable[ActorRef], failure: Throwable): Unit = {
actorInstrumentation(cell).processFailure(failure)
}
}
trait ActorInstrumentationAware {
def actorInstrumentation: ActorMonitor
def setActorInstrumentation(ai: ActorMonitor): Unit
}
object ActorInstrumentationAware {
def apply(): ActorInstrumentationAware = new ActorInstrumentationAware {
private var _ai: ActorMonitor = _
def setActorInstrumentation(ai: ActorMonitor): Unit = _ai = ai
def actorInstrumentation: ActorMonitor = _ai
}
}
@Aspect
class MetricsIntoActorCellsMixin {
@DeclareMixin("akka.actor.ActorCell")
def mixinActorCellMetricsToActorCell: ActorInstrumentationAware = ActorInstrumentationAware()
@DeclareMixin("akka.actor.UnstartedCell")
def mixinActorCellMetricsToUnstartedActorCell: ActorInstrumentationAware = ActorInstrumentationAware()
}
|
