如何记录Akka HTTP客户端请求

Question

我需要记录akka http客户端请求及其响应。虽然似乎有一些用于记录这些请求的API，但没有关于如何完成这些请求的明确文档。我的方法是创建一个记录的请求，透明地包装Http().singleRequest(req)如下：

def loggedRequest(req: HttpRequest)
                  (implicit system: ActorSystem, ctx: ExecutionContext, m: Materializer): Future[HttpResponse] = {

  Http().singleRequest(req).map { resp ⇒
    Unmarshal(resp.entity).to[String].foreach{s ⇒
      system.log.info(req.toString)
      system.log.info(resp.toString + "\n" + s)
    }
    resp
  }
}

不幸的是，我必须通过unmarshal或通过简单地请求resp.entity.dataBytes来获取未来，以便恢复响应的主体。我得到了日志记录，但承诺已完成，我无法再将实体解组为实际数据。一个工作的解决方案将记录请求和响应并传递此测试用例，而不会抛出带有“Promise already completed”的IllegalStateException：

describe("Logged rest requests") {

  it("deliver typed responses") {
    val foo = Rest.loggedRequest(Get(s"http://127.0.0.1:9000/some/path"))
    val resp = foo.futureValue(patience)
    resp.status shouldBe StatusCodes.OK
    val res = Unmarshal(resp.entity).to[MyClass].futureValue
  }
}

想法欢迎。

Answer 1

我找到的解决方案之一是使用：

import akka.http.scaladsl.server.directives.DebuggingDirectives

val clientRouteLogged = DebuggingDirectives.logRequestResult("Client ReST", Logging.InfoLevel)(clientRoute)
Http().bindAndHandle(clientRouteLogged, interface, port)

哪个可以轻松记录请求并导致原始（字节）格式。问题是这些日志完全不可读。这里是变得复杂的地方。

这是我的例子，它编码请求/响应的实体并将其写入记录器。

您可以将函数传递给：

DebuggingDirectives.logRequestResult

def logRequestResult(magnet: LoggingMagnet[HttpRequest ⇒ RouteResult ⇒ Unit])

这是使用magnet pattern编写的函数：

LoggingMagnet[HttpRequest ⇒ RouteResult ⇒ Unit]

哪里：

LoggingMagnet[T](f: LoggingAdapter ⇒ T)

多亏了我们可以访问记录请求和结果所需的所有部分。我们有LoggingAdapter，HttpRequest和RouteResult

在我的情况下，我创建了一个内部函数。我不想再次传递所有参数。

def logRequestResult(level: LogLevel, route: Route)
                      (implicit m: Materializer, ex: ExecutionContext) = {
  def myLoggingFunction(logger: LoggingAdapter)(req: HttpRequest)(res: Any): Unit = {
    val entry = res match {
      case Complete(resp) =>
        entityAsString(resp.entity).map(data ⇒ LogEntry(s"${req.method} ${req.uri}: ${resp.status} \n entity: $data", level))
      case other =>
        Future.successful(LogEntry(s"$other", level))
    }
    entry.map(_.logTo(logger))
  }
  DebuggingDirectives.logRequestResult(LoggingMagnet(log => myLoggingFunction(log)))(route)
}

最重要的部分是我将myLoggingFunction放入logRequestResult的最后一行。

名为myLoggingFunction的函数，简单地匹配服务器计算的结果，并基于它创建一个LogEntry。

最后一个是允许从流中解码结果实体的方法。

def entityAsString(entity: HttpEntity)
                   (implicit m: Materializer, ex: ExecutionContext): Future[String] = {
entity.dataBytes
  .map(_.decodeString(entity.contentType().charset().value))
  .runWith(Sink.head)
}

该方法可以轻松添加到任何akka-http路由。

val myLoggedRoute = logRequestResult(Logging.InfoLevel, clinetRoute)
Http().bindAndHandle(myLoggedRoute, interface, port)

Answer 2

对于另一个解决方案，此代码记录请求IP并将随机数与每个请求和响应相关联，以便它们可以在日志中关联。它还记录响应时间。

由于请求可能需要一段时间才能处理，并且可能会失败，因此我希望立即查看请求，并在返回时查看响应。

RequestFields只是我要求的数据。默认情况下会有很多噪音。

val logRequestResponse: Directive0 =
  extractRequestContext flatMap { ctx =>
    extractClientIP flatMap { ip =>
      val id = scala.math.abs(rand.nextLong).toString
      onSuccess(RequestFields.fromIdIpAndRequest(id, ip, ctx.request)) flatMap { req =>
        logger.info("request", req.asJson)
        val i = Instant.now()
        mapRouteResultWith { result => 
          Result.fromIdStartTimeAndRouteResult(id, i, result) map { res =>
            logger.info("response", res.asJson)
            result
        }
      }
    }
  }
}