/* * ========================================================================================= * Copyright © 2013 the kamon project * * 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.metric import akka.actor import kamon.metric.SubscriptionsDispatcher.{ Unsubscribe, Subscribe } import kamon.{ ModuleSupervisor, Kamon } import kamon.metric.instrument.{ InstrumentFactory, CollectionContext } import scala.collection.concurrent.TrieMap import akka.actor._ import kamon.util.{ FastDispatch, TriemapAtomicGetOrElseUpdate } object Metrics extends ExtensionId[MetricsExtension] with ExtensionIdProvider { override def get(system: ActorSystem): MetricsExtension = super.get(system) def lookup(): ExtensionId[_ <: actor.Extension] = Metrics def createExtension(system: ExtendedActorSystem): MetricsExtension = new MetricsExtensionImpl(system) } case class EntityRegistration[T <: EntityRecorder](entity: Entity, recorder: T) trait MetricsExtension extends Kamon.Extension { def settings: MetricsExtensionSettings def shouldTrack(entity: Entity): Boolean def shouldTrack(entityName: String, category: String): Boolean = shouldTrack(Entity(entityName, category)) def register[T <: EntityRecorder](recorderFactory: EntityRecorderFactory[T], entityName: String): Option[EntityRegistration[T]] def register[T <: EntityRecorder](entity: Entity, recorder: T): EntityRegistration[T] def unregister(entity: Entity): Unit def find(entity: Entity): Option[EntityRecorder] def find(name: String, category: String): Option[EntityRecorder] def subscribe(filter: SubscriptionFilter, subscriber: ActorRef): Unit = subscribe(filter, subscriber, permanently = false) def subscribe(category: String, selection: String, subscriber: ActorRef, permanently: Boolean): Unit = subscribe(SubscriptionFilter(category, selection), subscriber, permanently) def subscribe(category: String, selection: String, subscriber: ActorRef): Unit = subscribe(SubscriptionFilter(category, selection), subscriber, permanently = false) def subscribe(filter: SubscriptionFilter, subscriber: ActorRef, permanently: Boolean): Unit def unsubscribe(subscriber: ActorRef): Unit def buildDefaultCollectionContext: CollectionContext def instrumentFactory(category: String): InstrumentFactory } class MetricsExtensionImpl(system: ExtendedActorSystem) extends MetricsExtension { import FastDispatch.Syntax val settings = MetricsExtensionSettings(system) private val _trackedEntities = TrieMap.empty[Entity, EntityRecorder] private val _collectionContext = buildDefaultCollectionContext private val _metricsCollectionDispatcher = system.dispatchers.lookup(settings.metricCollectionDispatcher) private lazy val _subscriptions = ModuleSupervisor.get(system).createModule("subscriptions-dispatcher", SubscriptionsDispatcher.props(settings.tickInterval, collectSnapshots).withDispatcher(settings.metricCollectionDispatcher)) def shouldTrack(entity: Entity): Boolean = settings.entityFilters.get(entity.category).map { filter ⇒ filter.accept(entity.name) } getOrElse (settings.trackUnmatchedEntities) def register[T <: EntityRecorder](recorderFactory: EntityRecorderFactory[T], entityName: String): Option[EntityRegistration[T]] = { import TriemapAtomicGetOrElseUpdate.Syntax val entity = Entity(entityName, recorderFactory.category) if (shouldTrack(entity)) { val instrumentFactory = settings.instrumentFactories.get(recorderFactory.category).getOrElse(settings.defaultInstrumentFactory) val recorder = _trackedEntities.atomicGetOrElseUpdate(entity, recorderFactory.createRecorder(instrumentFactory)).asInstanceOf[T] Some(EntityRegistration(entity, recorder)) } else None } def register[T <: EntityRecorder](entity: Entity, recorder: T): EntityRegistration[T] = { import TriemapAtomicGetOrElseUpdate.Syntax EntityRegistration(entity, _trackedEntities.atomicGetOrElseUpdate(entity, recorder).asInstanceOf[T]) } def unregister(entity: Entity): Unit = _trackedEntities.remove(entity).map(_.cleanup) def find(entity: Entity): Option[EntityRecorder] = _trackedEntities.get(entity) def find(name: String, category: String): Option[EntityRecorder] = find(Entity(name, category)) def subscribe(filter: SubscriptionFilter, subscriber: ActorRef, permanent: Boolean): Unit = _subscriptions.fastDispatch(Subscribe(filter, subscriber, permanent))(_metricsCollectionDispatcher) def unsubscribe(subscriber: ActorRef): Unit = _subscriptions.fastDispatch(Unsubscribe(subscriber))(_metricsCollectionDispatcher) def buildDefaultCollectionContext: CollectionContext = CollectionContext(settings.defaultCollectionContextBufferSize) def instrumentFactory(category: String): InstrumentFactory = settings.instrumentFactories.getOrElse(category, settings.defaultInstrumentFactory) /** * Collect and dispatch. */ private def collectSnapshots(): Map[Entity, EntitySnapshot] = { val builder = Map.newBuilder[Entity, EntitySnapshot] _trackedEntities.foreach { case (identity, recorder) ⇒ builder += ((identity, recorder.collect(_collectionContext))) } builder.result() } /* def printInitializationMessage(eventStream: EventStream, disableWeaverMissingError: Boolean): Unit = { if (!disableWeaverMissingError) { val weaverMissingMessage = """ | | ___ _ ___ _ _ ___ ___ _ _ | / _ \ | | |_ | | | | | | \/ |(_) (_) |/ /_\ \ ___ _ __ ___ ___ | |_ | | | | | | ___ __ _ __ __ ___ _ __ | . . | _ ___ ___ _ _ __ __ _ || _ |/ __|| '_ \ / _ \ / __|| __| | | | |/\| | / _ \ / _` |\ \ / // _ \| '__| | |\/| || |/ __|/ __|| || '_ \ / _` | || | | |\__ \| |_) || __/| (__ | |_ /\__/ / \ /\ /| __/| (_| | \ V /| __/| | | | | || |\__ \\__ \| || | | || (_| | |\_| |_/|___/| .__/ \___| \___| \__|\____/ \/ \/ \___| \__,_| \_/ \___||_| \_| |_/|_||___/|___/|_||_| |_| \__, | | | | __/ | | |_| |___/ | | It seems like your application wasn't started with the -javaagent:/path-to-aspectj-weaver.jar option. Without that Kamon might | not work properly, if you need help on setting up the weaver go to http://kamon.io/introduction/get-started/ for more info. If | you are sure that you don't need the weaver (e.g. you are only using KamonStandalone) then you can disable this error message | by changing the kamon.metrics.disable-aspectj-weaver-missing-error setting in your configuration file. | """.stripMargin eventStream.publish(Error("MetricsExtension", classOf[MetricsExtension], weaverMissingMessage)) } }*/ }