First stab at using a cachedHostConnectionPool in a High level akka http DSL. Closes cromwell#2130 #3
Conversation
| implicit val materializer = ActorMaterializer() | ||
|
|
||
| val config = ConfigFactory.load() | ||
| // we shoudl be able to control this pool with lots of different configs talked about here. We use `Unit` here because | ||
| // there isnt a context we need to pass around to make sure resquests get the correct response because we only call the |
There was a problem hiding this comment.
isnt -> isn't
resquests -> requests
| implicit val materializer = ActorMaterializer() | ||
|
|
||
| val config = ConfigFactory.load() | ||
| // we shoudl be able to control this pool with lots of different configs talked about here. We use `Unit` here because |
There was a problem hiding this comment.
we -> We
shoudl -> should
| // we shoudl be able to control this pool with lots of different configs talked about here. We use `Unit` here because | ||
| // there isnt a context we need to pass around to make sure resquests get the correct response because we only call the | ||
| // pool using Source.single but its something we could use in the future if need be | ||
| val cromwellPool = Http().cachedHostConnectionPool[Unit](host = config.getString("cromwell.interface"), port = config.getInt("cromwell.port")) |
There was a problem hiding this comment.
These values are used a few times. Let's validate once and reuse the results
| val config: Config | ||
| val cromwellPool: Flow[(HttpRequest, Unit), (Try[HttpResponse], Unit), HostConnectionPool] | ||
|
|
||
| // Header that akka http DSL api adds to a request when we recieve, We get an warning in logs if we dont strip it out before sending the request to cromwell |
| } | ||
|
|
||
| // left in here for now in case we decide to not use akka streams to send requests in the future, can probably remove soon | ||
| def forwardToCromwellNoStreaming(httpRequest: HttpRequest) : Future[HttpResponse] = { |
There was a problem hiding this comment.
I agree about removing this ASAP if it won't be used
| } | ||
|
|
||
| def forwardToCromwellStreaming(httpRequest: HttpRequest) = { | ||
| Source.single((cromIamRequestToCromwellRequest(httpRequest), ())).via(cromwellPool).runWith(Sink.head).map(cromwellResponseHandler) |
There was a problem hiding this comment.
I'd suggest to use Http.singleRequest. It's faster and less complicated. Using the streaming pool interface makes only sense if you really supply it with a stream of requests. Otherwise, it's functionally equivalent to use Http.singleRequest.
cjllanwarne
changed the title
cjllanwarne
changed the title
| returnInternalServerError("engine stats") | ||
| extractRequest { req => | ||
| complete { | ||
| forwardToCromwell(req) |
There was a problem hiding this comment.
I don't think we want to forward any stats requests to Cromwell, we should return some kind of "no you cant do that" response directly from CromIam - broadinstitute/cromwell#2137.
There was a problem hiding this comment.
and ignore the typo from that geoffjentry fellow
There was a problem hiding this comment.
Oh right, I did wonder why that one was left unchanged!
|
|
||
| // Header that Akka HTTP adds to every request on receive. | ||
| // We get an warning in logs if we don't strip it out before sending the request to cromwell | ||
| // HTTP header ‘Timeout-Access: <function1>’ is not allowed in requests |
There was a problem hiding this comment.
Maybe add this link as further justification: "https://github.com/akka/akka-http/issues/64"
| extractRequest { req => | ||
| complete { | ||
| forwardToCromwell(req) | ||
| } |
There was a problem hiding this comment.
Is there any way to DRY this out? It seems very repetitive.
There was a problem hiding this comment.
Yeah, I thought about that... I'm gonna hold off on that until after we get the extra validation steps in, in case we end up with a premature abstraction that we just end up unpicking.
| val logger = NoLogging | ||
|
|
||
| "Stats endpoint" should "respond with internal server error" in { | ||
| override def cromwellInterface: String = ??? | ||
| override def cromwellPort: Int = ??? |
There was a problem hiding this comment.
??? is really meant as a WIP placeholder, these should probably be changed to throw NotImplementedErrors
|
👍 |
|
Should there be pullapprove on this project? |
|
From a CromIAM client perspective, the responses look like they come from Cromwell directly it seems. Is this the desired behavior ? Or will it be changed in a later PR ? |
cjllanwarne
force-pushed
the
js_akka_streams
branch
from
456b884 to
6ccc67f
Compare
…).singlerequest that talks to a local cromwell and responds to the user with that request
cjllanwarne
force-pushed
the
js_akka_streams
branch
from
6ccc67f to
f14d265
Compare
Wanted to try using a
cachedHostConnectionPoolwhile still making use of the high level dslRouterequest parsing. Everything functionally works but there are still things that need to be made better or just changed completely.