10

Akka ストリーム用のインストルメンテーションをセットアップしようとしています。うまくいきましたが、ストリームの一部であるすべてのフローに名前を付けましたが、メトリクスにはまだこのような名前が付けられています:flow-0-0-unknown-operation

私がやろうとしていることの簡単な例:

val myflow = Flow[String].named("myflow").map(println)

Source.via(myflow).to(Sink.ignore).run()

基本的に、「myflow」用に作成されたアクターのメトリックを適切な名前で表示したいと考えています。

これは可能ですか？何か不足していますか？

4

1 に答える 1

1

私は自分のプロジェクトでこの課題を抱えていましたが、Kamon + Prometheus を使用して解決しました。Flowただし、名前metricNameを設定し、.xml を使用してメトリクス値をエクスポートできるAkka Stream を作成する必要がありましたval kamonThroughputGauge: Metric.Gauge。

class MonitorProcessingTimerFlow[T](interval: FiniteDuration)(metricName: String = "monitorFlow") extends GraphStage[FlowShape[T, T]] {
  val in = Inlet[T]("MonitorProcessingTimerFlow.in")
  val out = Outlet[T]("MonitorProcessingTimerFlow.out")

  Kamon.init()
  val kamonThroughputGauge: Metric.Gauge = Kamon.gauge("akka-stream-throughput-monitor")
  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new TimerGraphStageLogic(shape) {
    // mutable state
    var open = false
    var count = 0
    var start = System.nanoTime
    setHandler(in, new InHandler {
      override def onPush(): Unit = {
        try {
          push(out, grab(in))
          count += 1
          if (!open) {
            open = true
            scheduleOnce(None, interval)
          }
        } catch {
          case e: Throwable => failStage(e)
        }
      }
    })
    setHandler(out, new OutHandler {
      override def onPull(): Unit = {
        pull(in)
      }
    })

    override protected def onTimer(timerKey: Any): Unit = {
      open = false
      val duration = (System.nanoTime - start) / 1e9d
      val throughput = count / duration
      kamonThroughputGauge.withTag("name", metricName).update(throughput)
      count = 0
      start = System.nanoTime
    }
  }
  override def shape: FlowShape[T, T] = FlowShape[T, T](in, out)
}

MonitorProcessingTimerFlow次に、 を使用してメトリックをエクスポートする単純なストリームを作成しました。

implicit val system = ActorSystem("FirstStreamMonitoring")
val source = Source(Stream.from(1)).throttle(1, 1 second)
/** Simulating workload fluctuation: A Flow that expand the event to a random number of multiple events */
val flow = Flow[Int].extrapolate { element =>
  Stream.continually(Random.nextInt(100)).take(Random.nextInt(100)).iterator
}
val monitorFlow = Flow.fromGraph(new MonitorProcessingTimerFlow[Int](5 seconds)("monitorFlow"))
val sink = Sink.foreach[Int](println)

val graph = source
  .via(flow)
  .via(monitorFlow)
  .to(sink)
graph.run()

での適切な構成application.conf:

kamon.instrumentation.akka.filters {
  actors.track {
    includes = [ "FirstStreamMonitoring/user/*" ]
  }
}

次の名前のプロメテウス コンソールでスループット メトリックを確認できますname="monitorFlow"。 ここに画像の説明を入力

于 2020-12-18T11:56:52.247 に答える