Skip to content

Added header when reporting failing invocations or initializations #116 #128

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 23 commits into from
Jun 17, 2020
Merged
Show file tree
Hide file tree
Changes from 15 commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Package.swift
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@ let package = Package(
]),
.testTarget(name: "AWSLambdaRuntimeCoreTests", dependencies: [
.byName(name: "AWSLambdaRuntimeCore"),
.product(name: "NIOTestUtils", package: "swift-nio"),
.product(name: "NIOFoundationCompat", package: "swift-nio"),
]),
.testTarget(name: "AWSLambdaRuntimeTests", dependencies: [
.byName(name: "AWSLambdaRuntimeCore"),
Expand Down
17 changes: 10 additions & 7 deletions Sources/AWSLambdaRuntimeCore/HTTPClient.swift
Original file line number Diff line number Diff line change
Expand Up @@ -43,13 +43,16 @@ internal final class HTTPClient {
timeout: timeout ?? self.configuration.requestTimeout))
}

func post(url: String, body: ByteBuffer?, timeout: TimeAmount? = nil) -> EventLoopFuture<Response> {
self.execute(Request(targetHost: self.targetHost,
url: url,
method: .POST,
headers: HTTPClient.headers,
body: body,
timeout: timeout ?? self.configuration.requestTimeout))
func post(url: String, body: ByteBuffer?, timeout: TimeAmount? = nil, additionalHeaders: HTTPHeaders? = nil) -> EventLoopFuture<Response> {
var headers = HTTPClient.headers
additionalHeaders.flatMap { headers.add(contentsOf: $0) }
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@tomerd I would vote to inject all headers on every call and to not have them split over two files. wdyt? In hindsight having static headers directly on the HTTPClient doesn't make much sense IMHO.

Copy link
Contributor

@tomerd tomerd Jun 16, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hi @fabianfett do you mean have RuntimeClient fully control the headers instead of injecting "additional" ones? if so, that sounds good to me.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it be ok to include that change in this PR or should I make a separate one for this?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I suggest we fix it in this PR

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok. I moved the static header declaration into an extension on RuntimeClient. The get and post methods on HTTPClient now have a non-optional headers parameter (without a default value). I also moved the headers parameter to come after the urlparameter ... that ordering made a more cohesive impression to me ^^


return self.execute(Request(targetHost: self.targetHost,
url: url,
method: .POST,
headers: headers,
body: body,
timeout: timeout ?? self.configuration.requestTimeout))
}

/// cancels the current request if there is one
Expand Down
2 changes: 1 addition & 1 deletion Sources/AWSLambdaRuntimeCore/LambdaConfiguration.swift
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ extension Lambda {
let requestTimeout: TimeAmount?

init(baseURL: String? = nil, keepAlive: Bool? = nil, requestTimeout: TimeAmount? = nil) {
let ipPort = env("AWS_LAMBDA_RUNTIME_API")?.split(separator: ":") ?? ["127.0.0.1", "7000"]
let ipPort = (baseURL ?? env("AWS_LAMBDA_RUNTIME_API"))?.split(separator: ":") ?? ["127.0.0.1", "7000"]
guard ipPort.count == 2, let port = Int(ipPort[1]) else {
preconditionFailure("invalid ip+port configuration \(ipPort)")
}
Expand Down
10 changes: 8 additions & 2 deletions Sources/AWSLambdaRuntimeCore/LambdaRuntimeClient.swift
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,9 @@ extension Lambda {
private let allocator = ByteBufferAllocator()
private let httpClient: HTTPClient

/// Headers that must be sent along an invocation or initialization error report
internal static let errorHeaders = HTTPHeaders([("lambda-runtime-function-error-type", "Unhandled")])

init(eventLoop: EventLoop, configuration: Configuration.RuntimeEngine) {
self.eventLoop = eventLoop
self.httpClient = HTTPClient(eventLoop: eventLoop, configuration: configuration)
Expand Down Expand Up @@ -61,6 +64,8 @@ extension Lambda {
func reportResults(logger: Logger, invocation: Invocation, result: Result<ByteBuffer?, Error>) -> EventLoopFuture<Void> {
var url = Consts.invocationURLPrefix + "/" + invocation.requestID
var body: ByteBuffer?
var additionalHeaders: HTTPHeaders?

switch result {
case .success(let buffer):
url += Consts.postResponseURLSuffix
Expand All @@ -71,9 +76,10 @@ extension Lambda {
let bytes = errorResponse.toJSONBytes()
body = self.allocator.buffer(capacity: bytes.count)
body!.writeBytes(bytes)
additionalHeaders = RuntimeClient.errorHeaders
}
logger.debug("reporting results to lambda runtime engine using \(url)")
return self.httpClient.post(url: url, body: body).flatMapThrowing { response in
return self.httpClient.post(url: url, body: body, additionalHeaders: additionalHeaders).flatMapThrowing { response in
guard response.status == .accepted else {
throw RuntimeError.badStatusCode(response.status)
}
Expand All @@ -98,7 +104,7 @@ extension Lambda {
var body = self.allocator.buffer(capacity: bytes.count)
body.writeBytes(bytes)
logger.warning("reporting initialization error to lambda runtime engine using \(url)")
return self.httpClient.post(url: url, body: body).flatMapThrowing { response in
return self.httpClient.post(url: url, body: body, additionalHeaders: RuntimeClient.errorHeaders).flatMapThrowing { response in
guard response.status == .accepted else {
throw RuntimeError.badStatusCode(response.status)
}
Expand Down
109 changes: 109 additions & 0 deletions Tests/AWSLambdaRuntimeCoreTests/LambdaRuntimeClientTest.swift
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,11 @@
//===----------------------------------------------------------------------===//

@testable import AWSLambdaRuntimeCore
import Logging
import NIO
import NIOFoundationCompat
@testable import NIOHTTP1
import NIOTestUtils
import XCTest

class LambdaRuntimeClientTest: XCTestCase {
Expand Down Expand Up @@ -209,6 +214,110 @@ class LambdaRuntimeClientTest: XCTestCase {
XCTAssertEqual(#"{"errorType":"error","errorMessage":"🥑👨‍👩‍👧‍👧👩‍👩‍👧‍👧👨‍👨‍👧"}"#, String(decoding: emojiBytes, as: Unicode.UTF8.self))
}

func testInitializationErrorReport() {
let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
defer { XCTAssertNoThrow(try eventLoopGroup.syncShutdownGracefully()) }

let server = NIOHTTP1TestServer(group: eventLoopGroup)
defer { XCTAssertNoThrow(try server.stop()) }

let logger = Logger(label: "TestLogger")
let client = Lambda.RuntimeClient(eventLoop: eventLoopGroup.next(), configuration: .init(baseURL: "127.0.0.1:\(server.serverPort)"))
let result = client.reportInitializationError(logger: logger, error: TestError("boom"))

var inboundHeader: HTTPServerRequestPart?
XCTAssertNoThrow(inboundHeader = try server.readInbound())
guard case .head(let head) = try? XCTUnwrap(inboundHeader) else { XCTFail("Expected to get a head first"); return }
XCTAssertEqual(head.headers["lambda-runtime-function-error-type"], ["Unhandled"])
XCTAssertEqual(head.headers["user-agent"], ["Swift-Lambda/Unknown"])

var inboundBody: HTTPServerRequestPart?
XCTAssertNoThrow(inboundBody = try server.readInbound())
guard case .body(let body) = try? XCTUnwrap(inboundBody) else { XCTFail("Expected body after head"); return }
XCTAssertEqual(try JSONDecoder().decode(ErrorResponse.self, from: body).errorMessage, "boom")

XCTAssertEqual(try server.readInbound(), .end(nil))

XCTAssertNoThrow(try server.writeOutbound(.head(.init(version: .init(major: 1, minor: 1), status: .accepted))))
XCTAssertNoThrow(try server.writeOutbound(.end(nil)))
XCTAssertNoThrow(try result.wait())
}

func testInvocationErrorReport() {
let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
defer { XCTAssertNoThrow(try eventLoopGroup.syncShutdownGracefully()) }

let server = NIOHTTP1TestServer(group: eventLoopGroup)
defer { XCTAssertNoThrow(try server.stop()) }

let logger = Logger(label: "TestLogger")
let client = Lambda.RuntimeClient(eventLoop: eventLoopGroup.next(), configuration: .init(baseURL: "127.0.0.1:\(server.serverPort)"))

let header = HTTPHeaders([
(AmazonHeaders.requestID, "test"),
(AmazonHeaders.deadline, String(Date(timeIntervalSinceNow: 60).millisSinceEpoch)),
(AmazonHeaders.invokedFunctionARN, "arn:aws:lambda:us-east-1:123456789012:function:custom-runtime"),
(AmazonHeaders.traceID, "Root=1-5bef4de7-ad49b0e87f6ef6c87fc2e700;Parent=9a9197af755a6419;Sampled=1"),
])
var inv: Lambda.Invocation?
XCTAssertNoThrow(inv = try Lambda.Invocation(headers: header))
guard let invocation = inv else { return }

let result = client.reportResults(logger: logger, invocation: invocation, result: Result.failure(TestError("boom")))

var inboundHeader: HTTPServerRequestPart?
XCTAssertNoThrow(inboundHeader = try server.readInbound())
guard case .head(let head) = try? XCTUnwrap(inboundHeader) else { XCTFail("Expected to get a head first"); return }
XCTAssertEqual(head.headers["lambda-runtime-function-error-type"], ["Unhandled"])
XCTAssertEqual(head.headers["user-agent"], ["Swift-Lambda/Unknown"])

var inboundBody: HTTPServerRequestPart?
XCTAssertNoThrow(inboundBody = try server.readInbound())
guard case .body(let body) = try? XCTUnwrap(inboundBody) else { XCTFail("Expected body after head"); return }
XCTAssertEqual(try JSONDecoder().decode(ErrorResponse.self, from: body).errorMessage, "boom")

XCTAssertEqual(try server.readInbound(), .end(nil))

XCTAssertNoThrow(try server.writeOutbound(.head(.init(version: .init(major: 1, minor: 1), status: .accepted))))
XCTAssertNoThrow(try server.writeOutbound(.end(nil)))
XCTAssertNoThrow(try result.wait())
}

func testSuccessHeaders() {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think I would rename this to testInvocationResponse because it isn't as generic as SuccessHeaders

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right. I think Headers is generally wrong at this point because the method is also testing a lot of other things as well.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is testInvocationSuccessResponseok with you as well? To disambiguate between success and error tests

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Ro-M yes testInvocationSuccessResponse is perfect

let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
defer { XCTAssertNoThrow(try eventLoopGroup.syncShutdownGracefully()) }

let server = NIOHTTP1TestServer(group: eventLoopGroup)
defer { XCTAssertNoThrow(try server.stop()) }

let logger = Logger(label: "TestLogger")
let client = Lambda.RuntimeClient(eventLoop: eventLoopGroup.next(), configuration: .init(baseURL: "127.0.0.1:\(server.serverPort)"))

let header = HTTPHeaders([
(AmazonHeaders.requestID, "test"),
(AmazonHeaders.deadline, String(Date(timeIntervalSinceNow: 60).millisSinceEpoch)),
(AmazonHeaders.invokedFunctionARN, "arn:aws:lambda:us-east-1:123456789012:function:custom-runtime"),
(AmazonHeaders.traceID, "Root=1-5bef4de7-ad49b0e87f6ef6c87fc2e700;Parent=9a9197af755a6419;Sampled=1"),
])
var inv: Lambda.Invocation?
XCTAssertNoThrow(inv = try Lambda.Invocation(headers: header))
guard let invocation = inv else { return }

let result = client.reportResults(logger: logger, invocation: invocation, result: Result.success(nil))

var inboundHeader: HTTPServerRequestPart?
XCTAssertNoThrow(inboundHeader = try server.readInbound())
guard case .head(let head) = try? XCTUnwrap(inboundHeader) else { XCTFail("Expected to get a head first"); return }
XCTAssertFalse(head.headers.contains(name: "lambda-runtime-function-error-type"))
XCTAssertEqual(head.headers["user-agent"], ["Swift-Lambda/Unknown"])

XCTAssertEqual(try server.readInbound(), .end(nil))

XCTAssertNoThrow(try server.writeOutbound(.head(.init(version: .init(major: 1, minor: 1), status: .accepted))))
XCTAssertNoThrow(try server.writeOutbound(.end(nil)))
XCTAssertNoThrow(try result.wait())
}

class Behavior: LambdaServerBehavior {
var state = 0

Expand Down