протокол низкоуровневой телеметрии GRPC - PullRequest
Новая
       36

протокол низкоуровневой телеметрии GRPC

0 голосов
/ 03 апреля 2019

Я пытаюсь измерить задержку моего обслуживания на более низком уровне.Оглядываясь вокруг, я увидел, что можно добавить addStreamTracerFactory в конструктор grpc.

Я сделал эту простую реализацию, подобную этой, и напечатал журналы: 

val server = io.grpc.netty.NettyServerBuilder.forPort(ApplicationConfig.Service.bindPort).addStreamTracerFactory(ServerStreamTracerFactory)....

class Telemetry(fullMethodName: String, headers: Metadata) extends ServerStreamTracer with LazyLogging {
  override def serverCallStarted(callInfo: ServerStreamTracer.ServerCallInfo[_, _]): Unit = {
    logger.info(s"Telemetry '$fullMethodName' '$headers' callinfo:$callInfo")
    super.serverCallStarted(callInfo)
  }

  override def inboundMessage(seqNo: Int): Unit = {
    logger.info(s"inboundMessage $seqNo")
    super.inboundMessage(seqNo)
  }
  override def inboundMessageRead(seqNo: Int, optionalWireSize: Long, optionalUncompressedSize: Long): Unit = {
    logger.info(s"inboundMessageRead $seqNo $optionalWireSize $optionalUncompressedSize")
    super.inboundMessageRead(seqNo, optionalWireSize, optionalUncompressedSize)
  }
  override def outboundMessage(seqNo: Int): Unit = {
    logger.info(s"outboundMessage $seqNo")
    super.outboundMessage(seqNo)
  }
  override def outboundMessageSent(seqNo: Int, optionalWireSize: Long, optionalUncompressedSize: Long): Unit = {
    logger.info(s"outboundMessageSent $seqNo $optionalWireSize $optionalUncompressedSize")
    super.outboundMessageSent(seqNo, optionalWireSize, optionalUncompressedSize)
  }
  override def streamClosed(status: Status): Unit = {
    logger.info(s"streamClosed $status")
    super.streamClosed(status)
  }

}

object ServerStreamTracerFactory extends Factory with LazyLogging{
  logger.info("called")
  override def newServerStreamTracer(fullMethodName: String, headers: Metadata): ServerStreamTracer = {
    logger.info(s"called with $fullMethodName $headers")
    new Telemetry(fullMethodName, headers)
  }
}

Я запускаю простой клиент grpc в цикле и проверяю вывод трассировки потока сервера.

Я вижу, что "жизненный цикл" журналов повторяется.Вот одна итерация (но она повторяет одно и то же снова и снова): 

22:15:06 INFO  [grpc-default-worker-ELG-3-2] [newServerStreamTracer:38] [ServerStreamTracerFactory$] called with com.dy.affinity.service.AffinityService/getAffinities Metadata(content-type=application/grpc,user-agent=grpc-python/1.15.0 grpc-c/6.0.0 (osx; chttp2; glider),grpc-accept-encoding=identity,deflate,gzip,accept-encoding=identity,gzip)
22:15:06 INFO  [grpc-default-executor-0] [serverCallStarted:8] [Telemetry] Telemetry 'com.dy.affinity.service.AffinityService/getAffinities' 'Metadata(content-type=application/grpc,user-agent=grpc-python/1.15.0 grpc-c/6.0.0 (osx; chttp2; glider),grpc-accept-encoding=identity,deflate,gzip,accept-encoding=identity,gzip)' callinfo:io.grpc.internal.ServerCallInfoImpl@5badffd8
22:15:06 INFO  [grpc-default-worker-ELG-3-2] [inboundMessage:13] [Telemetry] inboundMessage 0
22:15:06 INFO  [grpc-default-worker-ELG-3-2] [inboundMessageRead:17] [Telemetry] inboundMessageRead 0 19 -1
22:15:06 INFO  [pool-1-thread-5] [outboundMessage:21] [Telemetry] outboundMessage 0
22:15:06 INFO  [pool-1-thread-5] [outboundMessageSent:25] [Telemetry] outboundMessageSent 0 0 0
22:15:06 INFO  [grpc-default-worker-ELG-3-2] [streamClosed:29] [Telemetry] streamClosed Status{code=OK, description=null, cause=null}

Несколько вещей, которые мне не совсем понятны, если посмотреть на эти журналы:

  1. Почему новый поток создается для каждого запроса?Я думал, что клиент grpc должен повторно использовать соединение."поток закрыт" не должен называться правильно?
  2. Если поток используется повторно, почему я вижу, что число inboundMessage (и outboundMessage) всегда равно "0".(Также, когда я запустил несколько клиентов параллельно, это всегда 0).В каком случае номер сообщения не должен быть 0?
  3. Если поток не используется повторно, как мне настроить клиентов для повторного использования соединения?

1 Ответ

0 голосов
/ 05 апреля 2019

В gRPC для каждого RPC создается один поток HTTP2 (в то время как при повторных попытках или хеджировании может быть более одного потока для каждого RPC). Потоки HTTP2 мультиплексируются в одном соединении, и открывать и закрывать потоки довольно дешево. Таким образом, повторно используется соединение, а не поток.

Последовательность, которую вы получаете от методов трассировки, - это последовательность сообщений для этого потока, которая начинается с 0. Похоже, вы выполняете унарные RPC, которые делают один запрос и получают один ответ, а затем закрываются. То, что вы видите, совершенно нормально.

...