<commit_before>package oncue.svc.funnel
package agent
package jmx

import cjmx.util.jmx.{Attach, MBeanQuery, RichMBeanServerConnection, JMXConnection, JMX}
import journal.Logger
import scalaz.stream.{Process,Channel,io}
import scalaz.concurrent.{Task,Strategy}
import scalaz.std.option._
import scalaz.syntax.std.option._
import javax.management.remote.JMXConnector
import javax.management.{MBeanServerConnection, ObjectName, Attribute, MBeanInfo}
import java.util.concurrent.ConcurrentHashMap
import scala.collection.JavaConverters._
import scala.util.control.NonFatal
import scala.concurrent.duration._
import InstrumentKinds._

case class JMXImportException(override val getMessage: String) extends RuntimeException

object Import {
  import JMXConnection._
  import Monitoring.{serverPool,schedulingPool}

  private[this] val log = Logger[Import.type]

  type ConnectorCache = ConcurrentHashMap[String, JMXConnector]

  /**
   * Periodically fetch metrics from the specified JMX location.
   * See the configuration agent.cfg for details on specifiying queries
   * and attribute exclusion rules.
   */
  def periodicly(
    location: String,
    queries: Seq[MBeanQuery],
    exclusions: String => Boolean,
    cluster: String
  )(cache: ConnectorCache, inst: Instruments
  )(frequency: Duration = 10.seconds
  ): Process[Task,Unit] =
    Process.awakeEvery(frequency)(Strategy.Executor(serverPool), schedulingPool)
      .evalMap(_ => now(location, queries, exclusions, cluster)(cache, inst))<commit_msg>fix the spelling of Import.periodically<commit_after>package oncue.svc.funnel
package agent
package jmx

import cjmx.util.jmx.{Attach, MBeanQuery, RichMBeanServerConnection, JMXConnection, JMX}
import journal.Logger
import scalaz.stream.{Process,Channel,io}
import scalaz.concurrent.{Task,Strategy}
import scalaz.std.option._
import scalaz.syntax.std.option._
import javax.management.remote.JMXConnector
import javax.management.{MBeanServerConnection, ObjectName, Attribute, MBeanInfo}
import java.util.concurrent.ConcurrentHashMap
import scala.collection.JavaConverters._
import scala.util.control.NonFatal
import scala.concurrent.duration._
import InstrumentKinds._

case class JMXImportException(override val getMessage: String) extends RuntimeException

object Import {
  import JMXConnection._
  import Monitoring.{serverPool,schedulingPool}

  private[this] val log = Logger[Import.type]

  type ConnectorCache = ConcurrentHashMap[String, JMXConnector]

  /**
   * Periodically fetch metrics from the specified JMX location.
   * See the configuration agent.cfg for details on specifiying queries
   * and attribute exclusion rules.
   */
  def periodically(
    location: String,
    queries: Seq[MBeanQuery],
    exclusions: String => Boolean,
    cluster: String
  )(cache: ConnectorCache, inst: Instruments
  )(frequency: Duration = 10.seconds
  ): Process[Task,Unit] =
    Process.awakeEvery(frequency)(Strategy.Executor(serverPool), schedulingPool)
      .evalMap(_ => now(location, queries, exclusions, cluster)(cache, inst))
{'fraction_non_alphanumeric': 0.08956692913385826, 'fraction_numerical': 0.0013123359580052493, 'mean_word_length': 5.542918454935623, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4025004', 'n_tokens_mistral': 885, 'n_tokens_neox': 856, 'n_words': 221}