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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
|
/*
* =========================================================================================
* 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.{ ActorRef, ActorLogging, Actor }
import akka.event.LoggingAdapter
import akka.io.IO
import akka.util.Timeout
import com.typesafe.config.Config
import kamon.Kamon
import kamon.metric.{ SegmentMetrics, TraceMetrics, MetricsModule, TickMetricSnapshotBuffer }
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._
import scala.concurrent.duration.FiniteDuration
class Agent extends Actor with SprayJsonSupport with ActorLogging with MetricsSubscription {
import context.dispatcher
private val config = context.system.settings.config
val agentSettings = AgentSettings.fromConfig(config)
// Start the reporters
private val reporter = context.actorOf(MetricReporter.props(agentSettings), "metric-reporter")
val metricsSubscriber = {
val tickInterval = Kamon.metrics.settings.tickInterval
// Metrics are always sent to New Relic in 60 seconds intervals.
if (tickInterval == 60.seconds) reporter
else context.actorOf(TickMetricSnapshotBuffer.props(1 minute, reporter), "metric-buffer")
}
subscribeToMetrics(config, metricsSubscriber, Kamon.metrics)
// 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, scheme) ⇒ configureChildren(collector, runID, scheme)
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, scheme: String): Unit = {
log.info("Configuring New Relic reporters to use runID: [{}] and collector: [{}] over: [{}]", runID, collector, scheme)
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, scheme) ← connect(collector, agentSettings)
} yield Connected(collector, runID, scheme)) 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, String)] = {
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), apiClient.scheme)
}
}
}
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, scheme: String) 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, ssl: Boolean)
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!")
val ssl = newRelicConfig.getBoolean("ssl")
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,
ssl)
}
}
trait MetricsSubscription {
import kamon.util.ConfigTools.Syntax
import scala.collection.JavaConverters._
import MetricsSubscription._
def log: LoggingAdapter
def subscriptions(config: Config) = config getConfig "kamon.newrelic" getConfig "custom-metric-subscriptions"
def subscriptionKeys(config: Config) = subscriptions(config).firstLevelKeys filterNot isTraceOrSegmentEntityName
def subscribeToMetrics(config: Config, metricsSubscriber: ActorRef, extension: MetricsModule): Unit = {
subscribeToCustomMetrics(config, metricsSubscriber, extension)
subscribeToTransactionMetrics(metricsSubscriber, extension)
}
def subscribeToCustomMetrics(config: Config, metricsSubscriber: ActorRef, extension: MetricsModule): Unit =
subscriptionKeys(config) foreach { subscriptionCategory ⇒
subscriptions(config).getStringList(subscriptionCategory).asScala foreach { pattern ⇒
log.debug("Subscribing NewRelic reporting for custom metric '{}' : {}", subscriptionCategory, pattern)
extension.subscribe(subscriptionCategory, pattern, metricsSubscriber)
}
}
def subscribeToTransactionMetrics(metricsSubscriber: ActorRef, extension: MetricsModule): Unit =
traceAndSegmentMetrics foreach { subscriptionCategory ⇒
log.debug("Subscribing NewRelic reporting for transaction metric '{}' : {}", subscriptionCategory, defaultPattern)
extension.subscribe(subscriptionCategory, defaultPattern, metricsSubscriber)
}
}
object MetricsSubscription {
private val defaultPattern = "**"
private val traceAndSegmentMetrics = Seq(TraceMetrics.category, SegmentMetrics.category)
def isTraceOrSegmentEntityName(name: String): Boolean = traceAndSegmentMetrics contains name
}
|
