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
|
/*
* =========================================================================================
* Copyright © 2013-2014 the kamon project <http://kamon.io/>
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
* except in compliance with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the
* License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
* either express or implied. See the License for the specific language governing permissions
* and limitations under the License.
* =========================================================================================
*/
package kamon.newrelic
import akka.actor.{ ActorLogging, Actor }
import akka.io.IO
import akka.util.Timeout
import com.typesafe.config.Config
import spray.can.Http
import spray.json._
import scala.concurrent.Future
import spray.httpx.SprayJsonSupport
import spray.json.lenses.JsonLenses._
import java.lang.management.ManagementFactory
import kamon.util.ConfigTools.Syntax
import Agent._
import JsonProtocol._
import akka.pattern.pipe
import scala.concurrent.duration.FiniteDuration
class Agent extends Actor with SprayJsonSupport with ActorLogging {
import context.dispatcher
val agentSettings = AgentSettings.fromConfig(context.system.settings.config)
// Start the reporters
context.actorOf(MetricReporter.props(agentSettings), "metric-reporter")
// Start the connection to the New Relic collector.
self ! Connect
def receive: Receive = disconnected(agentSettings.maxConnectionRetries)
def disconnected(attemptsLeft: Int): Receive = {
case Connect ⇒ pipe(connectToCollector) to self
case Connected(collector, runID) ⇒ configureChildren(collector, runID)
case ConnectFailed(reason) if (attemptsLeft > 0) ⇒ scheduleReconnection(reason, attemptsLeft)
case ConnectFailed(reason) ⇒ giveUpConnection()
}
def connected: Receive = {
case Reconnect ⇒ reconnect()
case Shutdown ⇒ shutdown()
}
def reconnect(): Unit = {
log.warning("New Relic request the agent to restart the connection, all reporters will be paused until a new connection is available.")
self ! Connect
context.children.foreach(_ ! ResetConfiguration)
context become disconnected(agentSettings.maxConnectionRetries)
}
def shutdown(): Unit = {
log.error("New Relic requested the agent to be stopped, no metrics will be reported after this point.")
context stop self
}
def configureChildren(collector: String, runID: Long): Unit = {
log.info("Configuring New Relic reporters to use runID: [{}] and collector: [{}]", runID, collector)
context.children.foreach(_ ! Configure(collector, runID))
context become connected
}
def scheduleReconnection(connectionFailureReason: Throwable, attemptsLeft: Int): Unit = {
log.error(connectionFailureReason, "Initialization failed, retrying in {} seconds", agentSettings.retryDelay.toSeconds)
context.system.scheduler.scheduleOnce(agentSettings.retryDelay, self, Connect)
context become (disconnected(attemptsLeft - 1))
}
def giveUpConnection(): Unit = {
log.error("Giving up while trying to set up a connection with the New Relic collector. The New Relic module is shutting down itself.")
context.stop(self)
}
def connectToCollector: Future[ConnectResult] = {
(for {
collector ← selectCollector
runID ← connect(collector, agentSettings)
} yield Connected(collector, runID)) recover { case error ⇒ ConnectFailed(error) }
}
def selectCollector: Future[String] = {
val apiClient = new ApiMethodClient("collector.newrelic.com", None, agentSettings, IO(Http)(context.system))
apiClient.invokeMethod(RawMethods.GetRedirectHost, JsArray()) map { json ⇒
json.extract[String]('return_value)
}
}
def connect(collectorHost: String, connect: AgentSettings): Future[Long] = {
val apiClient = new ApiMethodClient(collectorHost, None, agentSettings, IO(Http)(context.system))
apiClient.invokeMethod(RawMethods.Connect, connect) map { json ⇒
json.extract[Long]('return_value / 'agent_run_id)
}
}
}
object Agent {
case object Connect
case object Reconnect
case object Shutdown
case object ResetConfiguration
case class Configure(collector: String, runID: Long)
sealed trait ConnectResult
case class Connected(collector: String, runID: Long) extends ConnectResult
case class ConnectFailed(reason: Throwable) extends ConnectResult
}
case class AgentSettings(licenseKey: String, appName: String, hostname: String, pid: Int, operationTimeout: Timeout,
maxConnectionRetries: Int, retryDelay: FiniteDuration, apdexT: Double)
object AgentSettings {
def fromConfig(config: Config) = {
// Name has the format of 'pid'@'host'
val runtimeName = ManagementFactory.getRuntimeMXBean.getName.split('@')
val newRelicConfig = config.getConfig("kamon.newrelic")
val licenseKey = newRelicConfig.getString("license-key")
assert(licenseKey != "<put-your-key-here>", "You forgot to include your New Relic license key in the configuration settings!")
AgentSettings(
licenseKey,
newRelicConfig.getString("app-name"),
runtimeName(1),
runtimeName(0).toInt,
Timeout(newRelicConfig.getFiniteDuration("operation-timeout")),
newRelicConfig.getInt("max-connect-retries"),
newRelicConfig.getFiniteDuration("connect-retry-delay"),
newRelicConfig.getFiniteDuration("apdexT").toMillis / 1E3D)
}
}
|
