diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md new file mode 100644 index 00000000..fdf17ad1 --- /dev/null +++ b/.github/PULL_REQUEST_TEMPLATE.md @@ -0,0 +1,33 @@ + + +#### What does this change do? + + +#### Why is it needed? What is the use case? + + +#### How did you test it? + + + + +#### Checklist + +- [ ] I have read [CONTRIBUTING.md](../CONTRIBUTING.md) and understand this pull request will be used + as a reference rather than merged directly +- [ ] I opened an issue first if this is a significant change +- [ ] This change is limited to the behaviour described above, with no unrelated reformatting +- [ ] Local tests pass through `make integ-tests-and-compile` + +By submitting this pull request, I confirm that my contribution is made under the terms of the +[Apache 2.0 license](https://www.apache.org/licenses/LICENSE-2.0). diff --git a/.github/workflows/check-binaries.yml b/.github/workflows/check-binaries.yml index 70b67dcd..320f54d4 100644 --- a/.github/workflows/check-binaries.yml +++ b/.github/workflows/check-binaries.yml @@ -23,11 +23,11 @@ jobs: with: ref: main - name: Download latest release - uses: robinraju/release-downloader@v1.13 - with: - latest: true - fileName: 'aws-lambda-rie*' - out-file-path: "bin" + run: | + mkdir -p bin + gh release download --pattern 'aws-lambda-rie*' --dir bin + env: + GH_TOKEN: ${{ github.token }} - name: Run check for vulnerabilities id: check-binaries run: | @@ -61,25 +61,29 @@ jobs: name: Save outputs for the check with the latest build id: save-new-version run: | - if [ "${{ steps.check-new-version.outcome }}" == "failure" ]; then + if [ "${CHECK_OUTCOME}" == "failure" ]; then fixed="No" else fixed="Yes" fi echo "fixed=$fixed" >> "$GITHUB_OUTPUT" + env: + CHECK_OUTCOME: ${{ steps.check-new-version.outcome }} - if: always() && steps.save-output.outputs.report_contents name: Create GitHub Issue indicating vulnerabilities id: create-issue - uses: dacbd/create-issue-action@main - with: - token: ${{ github.token }} - title: | - CVEs found in latest RIE release - body: | - ### CVEs found in latest RIE release - ``` - ${{ steps.save-output.outputs.report_contents }} - ``` - - #### Are these resolved by building with the latest patch version of Go (${{ steps.check-new-version.outputs.latest_version }})?: - > **${{ steps.save-new-version.outputs.fixed }}** + run: | + gh issue create \ + --title "CVEs found in latest RIE release" \ + --body "### CVEs found in latest RIE release + \`\`\` + ${REPORT_CONTENTS} + \`\`\` + + #### Are these resolved by building with the latest patch version of Go (${LATEST_VERSION})?: + > **${FIXED}**" + env: + GH_TOKEN: ${{ github.token }} + REPORT_CONTENTS: ${{ steps.save-output.outputs.report_contents }} + LATEST_VERSION: ${{ steps.check-new-version.outputs.latest_version }} + FIXED: ${{ steps.save-new-version.outputs.fixed }} diff --git a/.github/workflows/integ-tests.yml b/.github/workflows/integ-tests.yml index a87a8295..4e169758 100644 --- a/.github/workflows/integ-tests.yml +++ b/.github/workflows/integ-tests.yml @@ -8,16 +8,12 @@ name: Run Integration Tests jobs: go-tests: runs-on: ubuntu-latest - environment: - name: integ-tests steps: - uses: actions/checkout@v7 - name: run go tests run: make tests-with-docker integ-tests-x86: runs-on: ubuntu-latest - environment: - name: integ-tests steps: - uses: actions/checkout@v7 - uses: actions/setup-python@v7 @@ -27,8 +23,6 @@ jobs: run: make integ-tests-with-docker-x86-64 integ-tests-arm64: runs-on: ubuntu-latest - environment: - name: integ-tests steps: - uses: actions/checkout@v7 - uses: actions/setup-python@v7 @@ -38,8 +32,6 @@ jobs: run: make integ-tests-with-docker-arm64 integ-tests-old: runs-on: ubuntu-latest - environment: - name: integ-tests steps: - uses: actions/checkout@v7 - uses: actions/setup-python@v7 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index e681b226..f58e921c 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -33,12 +33,14 @@ jobs: make tests-with-docker make integ-tests - name: Release - uses: softprops/action-gh-release@v3 - with: - name: Release ${{ github.event.inputs.releaseVersion }} - tag_name: v${{ github.event.inputs.releaseVersion }} - body: ${{ github.event.inputs.releaseBody }} - files: | - bin/aws-lambda-rie - bin/aws-lambda-rie-arm64 + run: | + gh release create "v${RELEASE_VERSION}" \ + --title "Release ${RELEASE_VERSION}" \ + --notes "${RELEASE_BODY}" \ + bin/aws-lambda-rie \ + bin/aws-lambda-rie-arm64 \ bin/aws-lambda-rie-x86_64 + env: + GH_TOKEN: ${{ github.token }} + RELEASE_VERSION: ${{ github.event.inputs.releaseVersion }} + RELEASE_BODY: ${{ github.event.inputs.releaseBody }} diff --git a/.github/workflows/validate-branch-into-main.yaml b/.github/workflows/validate-branch-into-main.yaml index 77b2498b..336c5860 100644 --- a/.github/workflows/validate-branch-into-main.yaml +++ b/.github/workflows/validate-branch-into-main.yaml @@ -10,8 +10,9 @@ jobs: runs-on: ubuntu-latest steps: - name: Check source branch + env: + SOURCE_BRANCH: ${{ github.head_ref }} run: | - SOURCE_BRANCH="${{ github.head_ref }}" if [[ "$SOURCE_BRANCH" != "develop" ]]; then echo "Error: Only pull requests from develop branch are allowed into main" echo "Current source branch ($SOURCE_BRANCH)." diff --git a/.gitignore b/.gitignore index 6c886eba..7f80e8dc 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,4 @@ tags .idea .DS_Store .venv +build.log diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 5d48ddb4..29106ac0 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -6,10 +6,17 @@ documentation, we greatly value feedback and contributions from our community. Please read through this document before submitting any issues or pull requests to ensure we have all the necessary information to effectively respond to your bug report or contribution. +**Please note:** this repository is generated from an internal AWS source of truth, which means we are unable to merge +pull requests into it directly. We very much still want your input -- see +[How we handle contributions](#how-we-handle-contributions) for what happens to an issue or pull request you open, and +how you get credit for a change we adopt. + ## Reporting Bugs/Feature Requests -We welcome you to use the GitHub issue tracker to report bugs or suggest features. +We welcome you to use the GitHub issue tracker to report bugs or suggest features. Issues are the most effective way to +reach us: a feature request that describes your use case can be picked up and implemented, whereas a pull request cannot +be merged as-is. When filing an issue, please check existing open, or recently closed, issues to make sure somebody else hasn't already reported the issue. Please try to include as much information as you can. Details like these are incredibly useful: @@ -20,25 +27,34 @@ reported the issue. Please try to include as much information as you can. Detail * Anything unusual about your environment or deployment -## Contributing via Pull Requests -This repository contains the source code and examples for the Runtime Interface Emulator. We will accept pull requests on documentation, examples, bug fixes and the Dockerfiles. We will also accept pull requests, issues and feedback on improvements to the Runtime Interface Emulator. However, our priority will be to maintain fidelity with AWS Lambda’s Runtime Interface on the cloud. +## How we handle contributions + +This repository contains the source code and examples for the Runtime Interface Emulator. The code here is generated from +an internal AWS repository, which is the source of truth, and changes flow outward from there. Because of that we cannot +merge a pull request into this repository, even one we agree with. -Contributions via pull requests are much appreciated. Before sending us a pull request, please ensure that: +That does not mean we don't want it. Here is what each kind of contribution gets you: -1. You are working against the latest source on the *main* branch. -2. You check existing open, and recently merged, pull requests to make sure someone else hasn't addressed the problem already. -3. You open an issue to discuss any significant work - we would hate for your time to be wasted. +* **Feature requests and bug reports are welcome, and are the most useful thing you can send us.** Open an issue + describing the behaviour you want and the use case behind it. Our priority is maintaining fidelity with AWS Lambda's + Runtime Interface in the cloud, so a clear use case is what lets us weigh a change against that. +* **Pull requests are welcome as a reference.** We read them, and a working patch is often the clearest way to explain a + proposal. It will not be merged directly. If we adopt your solution, we make the corresponding change in the internal + repository, and it reaches this repository through the next sync. +* **If we adopt your change, we credit you in the release notes** for the version that ships it. -To send us a pull request, please: +So that we can act on a pull request, please: -1. Fork the repository. -2. Modify the source; please focus on the specific change you are contributing. If you also reformat all the code, it will be hard for us to focus on your change. -3. Ensure local tests pass through `make integ-tests-and-compile` -4. Commit to your fork using clear commit messages. -5. Send us a pull request, answering any default questions in the pull request interface. +1. Open an issue first to discuss any significant work -- we would hate for your time to be wasted on something we + cannot take. +2. Work against the latest source on the *main* branch. +3. Check existing open, and recently closed, pull requests and issues to make sure someone else hasn't raised it already. +4. Focus on the specific change you are proposing. If you also reformat all the code, it will be hard for us to see what + you are actually suggesting. +5. Ensure local tests pass through `make integ-tests-and-compile`. 6. Pay attention to any automated CI failures reported in the pull request, and stay involved in the conversation. -GitHub provides additional document on [forking a repository](https://help.github.com/articles/fork-a-repo/) and +GitHub provides additional documentation on [forking a repository](https://help.github.com/articles/fork-a-repo/) and [creating a pull request](https://help.github.com/articles/creating-a-pull-request/). @@ -58,4 +74,5 @@ If you discover a potential security issue in this project we ask that you notif ## Licensing -See the [LICENSE](LICENSE) file for our project's licensing. We will ask you to confirm the licensing of your contribution. +See the [LICENSE](LICENSE) file for our project's licensing. We will ask you to confirm the licensing of your +contribution, including for a pull request we adopt rather than merge. diff --git a/README.md b/README.md index a9345cdc..f39e227b 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,7 @@ Lambda’s orchestrator, or security and authentication configurations. You can * [To test an image without adding RIE to the image](#to-test-an-image-without-adding-rie-to-the-image) * [How to configure](#how-to-configure) * [Level of support](#level-of-support) +* [Contributing](#contributing) * [Security](#security) * [License](#license) @@ -187,6 +188,18 @@ configurations that will not be emulated by this component. * The component does _not_ support X-ray and other Lambda integrations locally. * The component supports only Linux, for x86-64 and arm64 architectures. +## Contributing + +We welcome feature requests and bug reports through the GitHub issue tracker, and they are the most effective way to +reach us. + +This repository is generated from an internal AWS source of truth, so we are unable to merge pull requests into it +directly. You are still welcome to open one: we read them, and a working patch is often the clearest way to explain a +proposal. It is treated as a reference rather than something we merge, and if we adopt your solution we make the change +in the internal repository and credit you in the release notes for the version that ships it. + +See [CONTRIBUTING](CONTRIBUTING.md) for details. + ## Security See [CONTRIBUTING](CONTRIBUTING.md#security-issue-notifications) for more information. diff --git a/internal/lambda-managed-instances/aws-lambda-rie/internal/app_test.go b/internal/lambda-managed-instances/aws-lambda-rie/internal/app_test.go index 31144c88..ed1d7420 100644 --- a/internal/lambda-managed-instances/aws-lambda-rie/internal/app_test.go +++ b/internal/lambda-managed-instances/aws-lambda-rie/internal/app_test.go @@ -79,7 +79,7 @@ func TestApp_ServeHTTP(t *testing.T) { defer mockApp.AssertExpectations(t) mockApp.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(tt.initResponse) if tt.initResponse == nil { - mockApp.On("Invoke", mock.Anything, mock.Anything, mock.Anything).Return(tt.invokeErr, tt.responseSent) + mockApp.On("Invoke", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(tt.invokeErr, tt.responseSent, false) } initMsg := intmodel.InitRequestMessage{ diff --git a/internal/lambda-managed-instances/aws-lambda-rie/internal/invoke/rie_invoke_request.go b/internal/lambda-managed-instances/aws-lambda-rie/internal/invoke/rie_invoke_request.go index 608d8e42..a5d1d3d1 100644 --- a/internal/lambda-managed-instances/aws-lambda-rie/internal/invoke/rie_invoke_request.go +++ b/internal/lambda-managed-instances/aws-lambda-rie/internal/invoke/rie_invoke_request.go @@ -40,6 +40,7 @@ type rieInvokeRequest struct { cognitoIdentityPoolId string clientContext string responseMode string + internalInvocationID string functionVersionID string } @@ -98,6 +99,7 @@ func NewRieInvokeRequest(request *http.Request, writer http.ResponseWriter) (*ri cognitoIdentityPoolId: cognitoIdentityPoolId, clientContext: clientContext, responseMode: request.Header.Get(invoke.ResponseModeHeader), + internalInvocationID: uuid.New().String(), } return req, nil @@ -184,3 +186,7 @@ func (r *rieInvokeRequest) UpdateFromInitData(initData interop.InitStaticDataPro func (r *rieInvokeRequest) FunctionVersionID() string { return r.functionVersionID } + +func (r *rieInvokeRequest) InternalInvocationID() string { + return r.internalInvocationID +} diff --git a/internal/lambda-managed-instances/aws-lambda-rie/internal/invoke/rie_invoke_request_test.go b/internal/lambda-managed-instances/aws-lambda-rie/internal/invoke/rie_invoke_request_test.go index 7ead546c..c4265441 100644 --- a/internal/lambda-managed-instances/aws-lambda-rie/internal/invoke/rie_invoke_request_test.go +++ b/internal/lambda-managed-instances/aws-lambda-rie/internal/invoke/rie_invoke_request_test.go @@ -143,6 +143,7 @@ func TestNewRieInvokeRequest(t *testing.T) { if tt.want.invokeID == "" { tt.want.invokeID = got.invokeID } + tt.want.internalInvocationID = got.internalInvocationID assert.Equal(t, tt.want, got) }) diff --git a/internal/lambda-managed-instances/aws-lambda-rie/internal/run.go b/internal/lambda-managed-instances/aws-lambda-rie/internal/run.go index 138b1038..2010ebb5 100644 --- a/internal/lambda-managed-instances/aws-lambda-rie/internal/run.go +++ b/internal/lambda-managed-instances/aws-lambda-rie/internal/run.go @@ -75,7 +75,7 @@ func Run(supv supvmodel.ProcessSupervisor, args []string, fileUtil utils.FileUti } rieApp := NewHTTPHandler(raptorApp, initMsg) - s, err := raptor.StartServer(raptorApp, rieApp, &raptor.TCPAddress{AddrPort: rieAddr}) + s, err := raptor.StartServer(raptorApp, rieApp, &raptor.TCPAddress{AddrPort: rieAddr}, false) if err != nil { return nil, nil, nil, fmt.Errorf("could not start RIE server: %w", err) } diff --git a/internal/lambda-managed-instances/aws-lambda-rie/internal/telemetry/README.md b/internal/lambda-managed-instances/aws-lambda-rie/internal/telemetry/README.md deleted file mode 100644 index 49f45e51..00000000 --- a/internal/lambda-managed-instances/aws-lambda-rie/internal/telemetry/README.md +++ /dev/null @@ -1,97 +0,0 @@ -# RIE Telemetry Package - -The RIE (Runtime Interface Emulator) telemetry package provides Telemetry API. - -## Architecture Overview - -``` -┌─────────────────┐ ┌─────────────────┐ ┌──────────────────┐ -│ EventsAPI │ │ LogsEgress │ │ SubscriptionAPI │ -│ │ │ │ │ │ -│ • Platform │ │ • Runtime logs │ │ • Subscription │ -│ events │ │ • Extension │ │ management │ -│ • Lifecycle │ │ logs │ │ • Schema │ -│ events │ │ • Log capture │ │ validation │ -└─────────┬───────┘ └─────────┬───────┘ └──────────┬───────┘ - │ │ │ - └──────────────┬───────────────────────────────┘ - │ - ┌────▼────┐ - │ Relay │ - │ │ - │ Event │ - │ Broker │ - └────┬────┘ - │ - ┌──────────────┼──────────────┐ - │ │ │ - ┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐ - │Subscriber │ │Subscriber │ │Subscriber │ - │ A │ │ B │ │ C │ - └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ - │ │ │ - ┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐ - │TCP Client │ │HTTP Client│ │TCP Client │ - └───────────┘ └───────────┘ └───────────┘ -``` - -## Core Components - -### 1. EventsAPI (`events_api.go`) -**Responsibility**: Platform event generation and distribution - -The EventsAPI serves as the primary interface for generating and broadcasting AWS Lambda platform events. It implements the `EventsAPI` interface and handles various lifecycle events including initialization, invocation, and error reporting. - -### 2. LogsEgress (`logs_egress.go`) -**Responsibility**: Log capture and forwarding - -The LogsEgress component implements the `StdLogsEgressAPI` interface to capture stdout/stderr from both runtime and extensions, forwarding them to telemetry subscribers while maintaining original console output. - -### 3. Relay (`relay.go`) -**Responsibility**: Event broadcasting and subscriber management - -The Relay acts as a central event broker, managing subscribers and broadcasting events to all registered telemetry consumers. - -### 4. SubscriptionAPI (`subscription_api.go`) -**Responsibility**: Subscription management and validation - -The SubscriptionAPI handles telemetry subscription requests, validates them against JSON schemas, and manages the subscription lifecycle. - -## Internal Components - -### 1. Subscriber (`internal/subscriber.go`) -**Responsibility**: Event batching and delivery - -Each subscriber represents a telemetry consumer and manages efficient event delivery through batching and asynchronous processing. - -### 2. Client (`internal/client.go`) -**Responsibility**: Protocol-specific event delivery - -The client abstraction provides protocol-specific implementations for delivering events to telemetry consumers. - -### 3. Batch (`internal/batch.go`) -**Responsibility**: Event collection and timing - -The batch component manages collections of events with size and time-based flushing logic. - -### 4. Types (`internal/types.go`) -**Responsibility**: Type definitions and constants - -Centralized type definitions for protocols, event categories, and configuration structures. - -## Event Flow - -### 1. Subscription Flow -``` -Extension/Agent → SubscriptionAPI → Schema Validation → Subscriber Creation → Relay Registration -``` - -### 2. Event Flow -``` -Event Source → EventsAPI → Relay → Subscribers → Batching → Client -``` - -### 3. Log Flow -``` -Runtime/Extension → LogsEgress → Console Output + Relay → Subscribers → Batching → Client -``` diff --git a/internal/lambda-managed-instances/aws-lambda-rie/main.go b/internal/lambda-managed-instances/aws-lambda-rie/main.go deleted file mode 100644 index 7f23d80e..00000000 --- a/internal/lambda-managed-instances/aws-lambda-rie/main.go +++ /dev/null @@ -1,12 +0,0 @@ -// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. -// SPDX-License-Identifier: Apache-2.0 - -package main - -import ( - "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda-managed-instances/aws-lambda-rie/run" -) - -func main() { - run.Run() -} diff --git a/internal/lambda-managed-instances/interop/mock_invoke_request.go b/internal/lambda-managed-instances/interop/mock_invoke_request.go index 84e0a1ac..b6aef276 100644 --- a/internal/lambda-managed-instances/interop/mock_invoke_request.go +++ b/internal/lambda-managed-instances/interop/mock_invoke_request.go @@ -143,6 +143,23 @@ func (_m *MockInvokeRequest) FunctionVersionID() string { return r0 } +func (_m *MockInvokeRequest) InternalInvocationID() string { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for InternalInvocationID") + } + + var r0 string + if rf, ok := ret.Get(0).(func() string); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(string) + } + + return r0 +} + func (_m *MockInvokeRequest) InvokeID() string { ret := _m.Called() diff --git a/internal/lambda-managed-instances/interop/sandbox_model.go b/internal/lambda-managed-instances/interop/sandbox_model.go index 2bb04530..8f50ea45 100644 --- a/internal/lambda-managed-instances/interop/sandbox_model.go +++ b/internal/lambda-managed-instances/interop/sandbox_model.go @@ -253,6 +253,7 @@ type EEStaticData struct { XRayTracingMode intmodel.XrayTracingMode ArtefactType intmodel.ArtefactType RuntimeVersion string + RuntimeRelease string AmiId string AvailabilityZoneId string } @@ -301,6 +302,8 @@ type InvokeRequest interface { UpdateFromInitData(InitStaticDataProvider) model.AppError FunctionVersionID() string + + InternalInvocationID() string } type InitStaticDataProvider interface { diff --git a/internal/lambda-managed-instances/interop/service_log_values.go b/internal/lambda-managed-instances/interop/service_log_values.go index cb6d1a5c..1ac339b6 100644 --- a/internal/lambda-managed-instances/interop/service_log_values.go +++ b/internal/lambda-managed-instances/interop/service_log_values.go @@ -27,6 +27,10 @@ const ( ShutdownWaitAllProcessesDuration = "WaitCustomerProcessesExitDuration" ShutdownRuntimeServerDuration = "StopRuntimeServerDuration" + RuntimeNextCountMetric = "RuntimeNextCount" + + RuntimeWorkerCountMetric = "RuntimeWorkerCount" + ClientErrorMetric = "ClientError" ClientErrorReasonTemplate = "ClientErrorReason-%s" CustomerErrorMetric = "CustomerError" diff --git a/internal/lambda-managed-instances/invoke/consts.go b/internal/lambda-managed-instances/invoke/consts.go index 1632a4f4..d6a140dc 100644 --- a/internal/lambda-managed-instances/invoke/consts.go +++ b/internal/lambda-managed-instances/invoke/consts.go @@ -9,6 +9,8 @@ const ( FunctionErrorBodyTrailer = "lambda-runtime-function-error-body" ResponseModeHeader = "invoke-response-mode" TraceIdHeader = "x-amzn-trace-id" + + RuntimeInvocationIdHeader = "lambda-runtime-invocation-id" ) type InvokeBodyResponseStatus string diff --git a/internal/lambda-managed-instances/invoke/invoke_router.go b/internal/lambda-managed-instances/invoke/invoke_router.go index 86a97d13..17be7f86 100644 --- a/internal/lambda-managed-instances/invoke/invoke_router.go +++ b/internal/lambda-managed-instances/invoke/invoke_router.go @@ -35,6 +35,8 @@ type RuntimeResponseRequest interface { BodyReader() io.Reader TrailerError() ErrorForInvoker + + InvocationID() string } type RuntimeErrorRequest interface { @@ -47,6 +49,8 @@ type RuntimeErrorRequest interface { ReturnCode() int ErrorDetails() string GetXrayErrorCause() json.RawMessage + + InvocationID() string } type runningInvoke interface { @@ -195,6 +199,12 @@ func (ir *InvokeRouter) GetRuntimePoolCounts() RuntimePoolCounts { func (ir *InvokeRouter) ReserveIdleRuntime(ctx context.Context, invokeID interop.InvokeID, timeout time.Duration) (interop.ReserveIdleRuntimeResponse, model.AppError) { logging.Debug(ctx, "InvokeRouter: reserving idle runtime") + if _, exists := ir.runningInvokes.Get(invokeID); exists { + logging.Warn(ctx, "InvokeRouter: reservation collides with in-flight invoke") + return interop.ReserveIdleRuntimeFailureResponse{ErrorType: model.ErrorDuplicatedInvokeId}, + model.NewClientError(ErrInvokeIdAlreadyExists, model.ErrorSeverityError, model.ErrorDuplicatedInvokeId) + } + err := ir.runtimePool.Reserve(invokeID, timeout, func() { if ir.runtimePool.ExpireReservation(invokeID) { logging.Info(ctx, "InvokeRouter: reservation expired") diff --git a/internal/lambda-managed-instances/invoke/invoke_router_test.go b/internal/lambda-managed-instances/invoke/invoke_router_test.go index c5353e8e..86685947 100644 --- a/internal/lambda-managed-instances/invoke/invoke_router_test.go +++ b/internal/lambda-managed-instances/invoke/invoke_router_test.go @@ -399,6 +399,29 @@ func TestReserveIdleRuntime_DuplicateInvokeID(t *testing.T) { assert.Equal(t, model.ErrorDuplicatedInvokeId, appErr.ErrorType()) } +func TestReserveIdleRuntime_DuplicateAgainstRunningInvoke(t *testing.T) { + t.Parallel() + + mocks, router := createMocksAndInitRouter() + + _, err := router.RuntimeNext(mocks.ctx, mocks.runtimeNextRequest) + require.NoError(t, err) + + const inFlightID = "reserve-vs-running" + router.runningInvokes.Set(inFlightID, &mocks.runnningInvoke) + + resp, appErr := router.ReserveIdleRuntime(mocks.ctx, inFlightID, 100*time.Millisecond) + + require.NotNil(t, appErr) + failResp, ok := resp.(interop.ReserveIdleRuntimeFailureResponse) + assert.True(t, ok, "expected ReserveIdleRuntimeFailureResponse") + assert.Equal(t, model.ErrorDuplicatedInvokeId, failResp.ErrorType) + assert.Equal(t, model.ErrorDuplicatedInvokeId, appErr.ErrorType()) + + assert.Equal(t, 0, router.runtimePool.ReservedCount(), "no reservation should have been recorded") + assert.Equal(t, 1, router.GetRuntimePoolCounts().Idle, "idle runtime should still be available") +} + func TestReserveIdleRuntime_Expiration(t *testing.T) { t.Parallel() diff --git a/internal/lambda-managed-instances/invoke/mock_runtime_error_request.go b/internal/lambda-managed-instances/invoke/mock_runtime_error_request.go index 09d09cf1..38b39284 100644 --- a/internal/lambda-managed-instances/invoke/mock_runtime_error_request.go +++ b/internal/lambda-managed-instances/invoke/mock_runtime_error_request.go @@ -171,6 +171,23 @@ func (_m *MockRuntimeErrorRequest) ReturnCode() int { return r0 } +func (_m *MockRuntimeErrorRequest) InvocationID() string { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for InvocationID") + } + + var r0 string + if rf, ok := ret.Get(0).(func() string); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(string) + } + + return r0 +} + func NewMockRuntimeErrorRequest(t interface { mock.TestingT Cleanup(func()) diff --git a/internal/lambda-managed-instances/invoke/mock_runtime_response_request.go b/internal/lambda-managed-instances/invoke/mock_runtime_response_request.go index 5da7b503..bc550bc8 100644 --- a/internal/lambda-managed-instances/invoke/mock_runtime_response_request.go +++ b/internal/lambda-managed-instances/invoke/mock_runtime_response_request.go @@ -122,6 +122,23 @@ func (_m *MockRuntimeResponseRequest) TrailerError() ErrorForInvoker { return r0 } +func (_m *MockRuntimeResponseRequest) InvocationID() string { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for InvocationID") + } + + var r0 string + if rf, ok := ret.Get(0).(func() string); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(string) + } + + return r0 +} + func NewMockRuntimeResponseRequest(t interface { mock.TestingT Cleanup(func()) diff --git a/internal/lambda-managed-instances/invoke/reservation_metrics.go b/internal/lambda-managed-instances/invoke/reservation_metrics.go index cd7446d4..57ef4963 100644 --- a/internal/lambda-managed-instances/invoke/reservation_metrics.go +++ b/internal/lambda-managed-instances/invoke/reservation_metrics.go @@ -13,11 +13,13 @@ import ( ) const ( - ReserveSuccessMetric = "ReserveSuccess" - ReserveFailedMetric = "ReserveFailed" + ReserveSuccessMetric = "ReserveSuccess" + ReserveFailedMetric = "ReserveFailed" + ReserveParseDurationMetric = "ReserveParseDuration" + ReserveLogicDurationMetric = "ReserveLogicDuration" ) -func ReserveServiceLog(logger servicelogs.Logger, opStart time.Time, invokeID string, appErr model.AppError) { +func ReserveServiceLog(logger servicelogs.Logger, opStart time.Time, invokeID string, appErr model.AppError, parseDuration, reserveDuration time.Duration) { props := []servicelogs.Property{ {Name: "invoke_id", Value: invokeID}, } @@ -43,6 +45,11 @@ func ReserveServiceLog(logger servicelogs.Logger, opStart time.Time, invokeID st servicelogs.Counter(ReserveFailedMetric, failed), servicelogs.Counter(interop.ClientErrorMetric, clientErrCnt), servicelogs.Counter(interop.NonCustomerErrorMetric, nonCustomerErrCnt), + servicelogs.Timer(ReserveParseDurationMetric, parseDuration), + } + + if reserveDuration > 0 { + metrics = append(metrics, servicelogs.Timer(ReserveLogicDurationMetric, reserveDuration)) } if appErr != nil { diff --git a/internal/lambda-managed-instances/invoke/running_invoke.go b/internal/lambda-managed-instances/invoke/running_invoke.go index feec3496..e08b12c7 100644 --- a/internal/lambda-managed-instances/invoke/running_invoke.go +++ b/internal/lambda-managed-instances/invoke/running_invoke.go @@ -68,6 +68,8 @@ type runningInvokeImpl struct { invokeRespSender InvokeResponseSender runtimeNext http.ResponseWriter + internalInvocationID string + responderFactoryFunc ResponderFactoryFunc sendInvokeToRuntime func(context.Context, interop.InitStaticDataProvider, interop.InvokeRequest, http.ResponseWriter, string) (int64, time.Duration, time.Duration, model.AppError) createTracingData func(traceId string, tracingMode intmodel.XrayTracingMode, segmentIDGenerator func() string) (downstreamTraceId string, tracingCtx *interop.TracingCtx) @@ -99,6 +101,8 @@ func newRunningInvoke( func (r *runningInvokeImpl) RunInvokeAndSendResult(ctx context.Context, initData interop.InitStaticDataProvider, invokeReq interop.InvokeRequest, metrics interop.InvokeMetrics) model.AppError { downstreamTraceId, tracingCtx := r.createTracingData(invokeReq.TraceId(), initData.XRayTracingMode(), xray.GenerateSegmentID) + r.internalInvocationID = invokeReq.InternalInvocationID() + metrics.TriggerStartRequest() if err := metrics.SendInvokeStartEvent(tracingCtx); err != nil { logging.Error(ctx, "Failed to send InvokeStartEvent", "err", err) @@ -290,11 +294,28 @@ func (r *runningInvokeImpl) RuntimeResponse(ctx context.Context, runtimeRespReq return model.NewCustomerError(model.ErrorRuntimeInvokeResponseInProgress) } + if echoedID := runtimeRespReq.InvocationID(); echoedID != "" && r.internalInvocationID != "" { + if echoedID != r.internalInvocationID { + logging.Warn(ctx, "Cross-wiring detected: invocation ID mismatch on response", + "expected", r.internalInvocationID, "received", echoedID) + r.responseState.CompareAndSwap(stateGotResponse, stateNoResponse) + return model.NewCustomerError(model.ErrorRuntimeInvokeTimeout) + } + } + r.runtimeResponseChan <- runtimeRespReq return <-r.responseSentChan } func (r *runningInvokeImpl) RuntimeError(ctx context.Context, runtimeErrReq RuntimeErrorRequest) model.AppError { + if echoedID := runtimeErrReq.InvocationID(); echoedID != "" && r.internalInvocationID != "" { + if echoedID != r.internalInvocationID { + logging.Warn(ctx, "Cross-wiring detected: invocation ID mismatch on error", + "expected", r.internalInvocationID, "received", echoedID) + return model.NewCustomerError(model.ErrorRuntimeInvokeTimeout) + } + } + oldState := r.responseState.Swap(stateGotError) if oldState == stateGotError { logging.Warn(ctx, "Invalid invoke state : error in progress") diff --git a/internal/lambda-managed-instances/invoke/running_invoke_test.go b/internal/lambda-managed-instances/invoke/running_invoke_test.go index 951f650e..4f69ae68 100644 --- a/internal/lambda-managed-instances/invoke/running_invoke_test.go +++ b/internal/lambda-managed-instances/invoke/running_invoke_test.go @@ -66,6 +66,10 @@ func hijackRunningInvokeDeps(ri *runningInvokeImpl, mocks *runningInvokeMocks) { func createMocksAndInitRunningInvoke(t *testing.T) (*runningInvokeMocks, *runningInvokeImpl) { mocks := newRunningInvokeMocks(t) + mocks.eaInvokeRequest.On("InternalInvocationID").Return("").Maybe() + mocks.runtimeRespReq.On("InvocationID").Return("").Maybe() + mocks.runtimeErrorReq.On("InvocationID").Return("").Maybe() + ri := newRunningInvoke( mocks.runtimeNextRequest, func(ctx context.Context, ir interop.InvokeRequest) InvokeResponseSender { diff --git a/internal/lambda-managed-instances/invoke/runtime_error_request.go b/internal/lambda-managed-instances/invoke/runtime_error_request.go index afe5edcd..990bf105 100644 --- a/internal/lambda-managed-instances/invoke/runtime_error_request.go +++ b/internal/lambda-managed-instances/invoke/runtime_error_request.go @@ -29,6 +29,7 @@ type runtimeError struct { invokeID interop.InvokeID errorType model.ErrorType errorCategory model.ErrorCategory + invocationID string errorDetails string xrayErrorCause json.RawMessage @@ -42,6 +43,7 @@ func NewRuntimeError(ctx context.Context, request *http.Request, invokeID intero invokeID: invokeID, errorType: model.GetValidRuntimeOrFunctionErrorType(request.Header.Get(RuntimeErrorTypeHeader)), errorCategory: RuntimeErrorCategory, + invocationID: request.Header.Get(RuntimeInvocationIdHeader), errorDetails: errorDetails, xrayErrorCause: getValidatedErrorCause(ctx, request.Header.Get(LambdaXRayErrorCauseHeader)), } @@ -89,6 +91,10 @@ func (r *runtimeError) GetXrayErrorCause() json.RawMessage { return r.xrayErrorCause } +func (r *runtimeError) InvocationID() string { + return r.invocationID +} + func (r *runtimeError) ReturnCode() int { return http.StatusOK } diff --git a/internal/lambda-managed-instances/invoke/runtime_pool.go b/internal/lambda-managed-instances/invoke/runtime_pool.go index df96162c..981d6972 100644 --- a/internal/lambda-managed-instances/invoke/runtime_pool.go +++ b/internal/lambda-managed-instances/invoke/runtime_pool.go @@ -5,6 +5,7 @@ package invoke import ( "errors" + "log/slog" "sync" "time" @@ -44,9 +45,14 @@ func (p *RuntimePool) Add(runtime runningInvoke) error { } func (p *RuntimePool) Reserve(invokeID interop.InvokeID, timeout time.Duration, onExpire func()) error { + lockStart := time.Now() p.mu.Lock() defer p.mu.Unlock() + if lockWait := time.Since(lockStart); lockWait > time.Millisecond { + slog.Warn("RuntimePool.Reserve lock contention", "wait_us", lockWait.Microseconds()) + } + if _, exists := p.reserved[invokeID]; exists { return ErrInvokeIdAlreadyExists } diff --git a/internal/lambda-managed-instances/invoke/runtime_response_request.go b/internal/lambda-managed-instances/invoke/runtime_response_request.go index b764af6c..7e51d52a 100644 --- a/internal/lambda-managed-instances/invoke/runtime_response_request.go +++ b/internal/lambda-managed-instances/invoke/runtime_response_request.go @@ -9,6 +9,7 @@ import ( "io" "log/slog" "net/http" + "time" "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda-managed-instances/interop" "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda-managed-instances/logging" @@ -23,24 +24,28 @@ const ( type runtimeResponse struct { request *http.Request + rc *http.ResponseController parsingErr model.AppError contentType string invokeID interop.InvokeID responseMode string + invocationID string } -func NewRuntimeResponse(ctx context.Context, request *http.Request, invokeID interop.InvokeID) runtimeResponse { +func NewRuntimeResponse(ctx context.Context, request *http.Request, writer http.ResponseWriter, invokeID interop.InvokeID) runtimeResponse { contentType := request.Header.Get(RuntimeContentTypeHeader) if contentType == "" { contentType = "application/octet-stream" } resp := runtimeResponse{ - request: request, - contentType: contentType, - invokeID: invokeID, + request: request, + rc: http.NewResponseController(writer), + contentType: contentType, + invokeID: invokeID, + invocationID: request.Header.Get(RuntimeInvocationIdHeader), } switch mode := request.Header.Get(RuntimeResponseModeHeader); mode { @@ -74,10 +79,20 @@ func (r *runtimeResponse) BodyReader() io.Reader { return r.request.Body } +func (r *runtimeResponse) Cancel() { + if err := r.rc.SetReadDeadline(time.Unix(0, 0)); err != nil { + slog.Warn("Cancel: SetReadDeadline failed", "err", err) + } +} + func (r *runtimeResponse) ResponseMode() string { return r.responseMode } +func (r *runtimeResponse) InvocationID() string { + return r.invocationID +} + func (r *runtimeResponse) TrailerError() ErrorForInvoker { typ := r.request.Trailer.Get(FunctionErrorTypeTrailer) if typ == "" { diff --git a/internal/lambda-managed-instances/invoke/runtime_response_request_test.go b/internal/lambda-managed-instances/invoke/runtime_response_request_test.go index cb11bf6f..a0c6e968 100644 --- a/internal/lambda-managed-instances/invoke/runtime_response_request_test.go +++ b/internal/lambda-managed-instances/invoke/runtime_response_request_test.go @@ -5,6 +5,7 @@ package invoke import ( "net/http" + "net/http/httptest" "testing" "github.com/stretchr/testify/assert" @@ -66,7 +67,7 @@ func TestRuntimeResponse_TrailerError(t *testing.T) { req.Trailer = make(http.Header) req.Trailer.Set(FunctionErrorTypeTrailer, tc.errorTypeTrailer) req.Trailer.Set(FunctionErrorBodyTrailer, tc.errorBodyTrailer) - resp := NewRuntimeResponse(req.Context(), req, "test-invoke-id") + resp := NewRuntimeResponse(req.Context(), req, httptest.NewRecorder(), "test-invoke-id") actualTrailerError := resp.TrailerError() diff --git a/internal/lambda-managed-instances/invoke/runtime_response_sender.go b/internal/lambda-managed-instances/invoke/runtime_response_sender.go index 9604a8dd..88a8d72b 100644 --- a/internal/lambda-managed-instances/invoke/runtime_response_sender.go +++ b/internal/lambda-managed-instances/invoke/runtime_response_sender.go @@ -33,9 +33,18 @@ func sendInvokeToRuntime(ctx context.Context, initData interop.InitStaticDataPro runtimeReq.Header().Set(RuntimeRequestIdHeader, invokeReq.InvokeID()) runtimeReq.Header().Set(RuntimeDeadlineHeader, strconv.FormatInt(invokeReq.Deadline().UnixMilli(), 10)) runtimeReq.Header().Set(RuntimeFunctionArnHeader, initData.FunctionARN()) - runtimeReq.Header().Set(RuntimeTraceIdHeader, traceId) - runtimeReq.Header().Set(RuntimeClientContextHeader, invokeReq.ClientContext()) - runtimeReq.Header().Set(RuntimeCognitoIdentifyHeader, buildCognitoIdentifyHeader(invokeReq)) + if traceId != "" { + runtimeReq.Header().Set(RuntimeTraceIdHeader, traceId) + } + if cc := invokeReq.ClientContext(); cc != "" { + runtimeReq.Header().Set(RuntimeClientContextHeader, cc) + } + if cogId := buildCognitoIdentifyHeader(invokeReq); cogId != "" { + runtimeReq.Header().Set(RuntimeCognitoIdentifyHeader, cogId) + } + if internalInvocationID := invokeReq.InternalInvocationID(); internalInvocationID != "" { + runtimeReq.Header().Set(RuntimeInvocationIdHeader, internalInvocationID) + } runtimeReq.WriteHeader(http.StatusOK) timedReader := &utils.TimedReader{ diff --git a/internal/lambda-managed-instances/invoke/runtime_response_sender_test.go b/internal/lambda-managed-instances/invoke/runtime_response_sender_test.go index 9727e5bc..a8796876 100644 --- a/internal/lambda-managed-instances/invoke/runtime_response_sender_test.go +++ b/internal/lambda-managed-instances/invoke/runtime_response_sender_test.go @@ -62,6 +62,7 @@ func buildInvokeReqMocks(invokeReq *interop.MockInvokeRequest) { invokeReq.On("ClientContext").Return("client-context-example") invokeReq.On("CognitoId").Return("cognito_id_12345") invokeReq.On("CognitoPoolId").Return("cognito_pool_id_6789") + invokeReq.On("InternalInvocationID").Return("") } func buildInitDataMocks(initData *interop.MockInitStaticDataProvider) { @@ -134,3 +135,37 @@ func TestSendResponseFailure_CtxCancelled(t *testing.T) { checkResponseSenderExpectations(t, mocks) } + +func TestSendResponse_OmitsEmptyOptionalHeaders(t *testing.T) { + t.Parallel() + + mocks := createMocksAndRuntimeResponder() + buildInitDataMocks(&mocks.initData) + + mocks.invokeReq.On("ContentType").Return("application/json") + mocks.invokeReq.On("InvokeID").Return("123456") + mocks.invokeReq.On("Deadline").Return(time.Now().Add(time.Second)) + mocks.invokeReq.On("ClientContext").Return("") + mocks.invokeReq.On("CognitoId").Return("") + mocks.invokeReq.On("CognitoPoolId").Return("") + mocks.invokeReq.On("InternalInvocationID").Return("") + mocks.invokeReq.On("BodyReader").Return(mocks.reader) + + recorder := httptest.NewRecorder() + mocks.runtimeReq = recorder + + _, _, _, err := sendInvokeToRuntime(mocks.ctx, &mocks.initData, &mocks.invokeReq, mocks.runtimeReq, "") + assert.NoError(t, err) + + headers := recorder.Header() + assert.Empty(t, headers.Get(RuntimeTraceIdHeader)) + assert.Empty(t, headers.Get(RuntimeClientContextHeader)) + assert.Empty(t, headers.Get(RuntimeCognitoIdentifyHeader)) + + assert.NotEmpty(t, headers.Get(RuntimeRequestIdHeader)) + assert.NotEmpty(t, headers.Get(RuntimeDeadlineHeader)) + assert.NotEmpty(t, headers.Get(RuntimeFunctionArnHeader)) + assert.NotEmpty(t, headers.Get(RuntimeContentTypeHeader)) + + checkResponseSenderExpectations(t, mocks) +} diff --git a/internal/lambda-managed-instances/rapi/handler/invocationresponse.go b/internal/lambda-managed-instances/rapi/handler/invocationresponse.go index 8fb6ce28..661d0d45 100644 --- a/internal/lambda-managed-instances/rapi/handler/invocationresponse.go +++ b/internal/lambda-managed-instances/rapi/handler/invocationresponse.go @@ -28,7 +28,7 @@ func (h *invocationResponseHandler) ServeHTTP(writer http.ResponseWriter, reques ctx := logging.WithInvokeID(request.Context(), invokeID) logging.Debug(ctx, "Received Runtime Response") - resp := invoke.NewRuntimeResponse(ctx, request, invokeID) + resp := invoke.NewRuntimeResponse(ctx, request, writer, invokeID) err := h.runtimeRespHandler.RuntimeResponse(ctx, &resp) if err == nil { diff --git a/internal/lambda-managed-instances/rapid/handlers.go b/internal/lambda-managed-instances/rapid/handlers.go index 5da1a59c..44fb6b94 100644 --- a/internal/lambda-managed-instances/rapid/handlers.go +++ b/internal/lambda-managed-instances/rapid/handlers.go @@ -346,6 +346,7 @@ func doInitRuntime( return customerErr } + execCtx.initExecutionData.StaticData.RuntimeRelease = appctx.GetRuntimeRelease(execCtx.appCtx) telemetry.SendInitRuntimeDoneLogEvent(execCtx.eventsAPI, execCtx.appCtx, phase, nil) return nil diff --git a/internal/lambda-managed-instances/rapid/model/client_error.go b/internal/lambda-managed-instances/rapid/model/client_error.go index 57802779..f2c8e5e2 100644 --- a/internal/lambda-managed-instances/rapid/model/client_error.go +++ b/internal/lambda-managed-instances/rapid/model/client_error.go @@ -17,6 +17,11 @@ const ( ErrorInvalidResponseBandwidthRate ErrorType = "ErrInvalidResponseBandwidthRate" ErrorInvalidResponseBandwidthBurstSize ErrorType = "ErrInvalidResponseBandwidthBurstSize" ErrorExecutionEnvironmentShutdown ErrorType = "Client.ExecutionEnvironmentShutDown" + + ErrorInvalidInvokeId ErrorType = "Client.InvalidInvokeId" + + ErrorInvalidConnectionHoldTimeout ErrorType = "ErrInvalidConnectionHoldTimeout" + ErrorInvalidResponseHoldTimeout ErrorType = "ErrInvalidResponseHoldTimeout" ) type ClientError struct { diff --git a/internal/lambda-managed-instances/rapid/model/function_metadata.go b/internal/lambda-managed-instances/rapid/model/function_metadata.go index 8349b7fe..f0147d02 100644 --- a/internal/lambda-managed-instances/rapid/model/function_metadata.go +++ b/internal/lambda-managed-instances/rapid/model/function_metadata.go @@ -4,12 +4,13 @@ package model type FunctionMetadata struct { - AccountID string - FunctionName string - FunctionVersion string - MemorySizeBytes uint64 - Handler string - RuntimeInfo RuntimeInfo + AccountID string + FunctionName string + FunctionVersion string + MemorySizeBytes uint64 + Handler string + RuntimeInfo RuntimeInfo + RuntimeWorkerCount int } type RuntimeInfo struct { diff --git a/internal/lambda-managed-instances/rapid/model/platform_error.go b/internal/lambda-managed-instances/rapid/model/platform_error.go index 3f67bc6e..2a787910 100644 --- a/internal/lambda-managed-instances/rapid/model/platform_error.go +++ b/internal/lambda-managed-instances/rapid/model/platform_error.go @@ -20,7 +20,8 @@ const ( ErrSandboxLogSocketsUnavailable ErrorType = "Sandbox.LogSocketsUnavailable" ErrSandboxEventSetupFailure ErrorType = "Sandbox.EventSetupFailure" - ErrSandboxShutdownFailed ErrorType = "Sandbox.ShutdownFailed" + ErrSandboxShutdownFailed ErrorType = "Sandbox.ShutdownFailed" + ErrorResponseReplayFailed ErrorType = "Sandbox.ResponseReplayFailed" ) type PlatformError struct { diff --git a/internal/lambda-managed-instances/raptor/raptor_utils.go b/internal/lambda-managed-instances/raptor/raptor_utils.go index dfdb4f40..3ea9b6ad 100644 --- a/internal/lambda-managed-instances/raptor/raptor_utils.go +++ b/internal/lambda-managed-instances/raptor/raptor_utils.go @@ -43,11 +43,12 @@ func getInitExecutionData(initRequest *internalModel.InitRequestMessage, runtime LogGroupName: initRequest.LogGroupName, LogStreamName: initRequest.LogStreamName, FunctionMetadata: model.FunctionMetadata{ - AccountID: initRequest.AccountID, - FunctionName: initRequest.TaskName, - FunctionVersion: initRequest.FunctionVersion, - MemorySizeBytes: uint64(initRequest.MemorySizeBytes), - Handler: initRequest.Handler, + AccountID: initRequest.AccountID, + FunctionName: initRequest.TaskName, + FunctionVersion: initRequest.FunctionVersion, + MemorySizeBytes: uint64(initRequest.MemorySizeBytes), + Handler: initRequest.Handler, + RuntimeWorkerCount: initRequest.RuntimeWorkerCount, RuntimeInfo: model.RuntimeInfo{ Arn: initRequest.RuntimeArn, Version: initRequest.RuntimeVersion, diff --git a/internal/lambda-managed-instances/raptor/server.go b/internal/lambda-managed-instances/raptor/server.go index 23374f88..f2b46376 100644 --- a/internal/lambda-managed-instances/raptor/server.go +++ b/internal/lambda-managed-instances/raptor/server.go @@ -4,6 +4,7 @@ package raptor import ( + "context" "log/slog" "net" "net/http" @@ -18,7 +19,7 @@ import ( "github.com/aws/aws-lambda-runtime-interface-emulator/internal/lambda-managed-instances/rapid/model" ) -func StartServer(shutdownHandler shutdownHandler, handler http.Handler, addr Address) (*Server, error) { +func StartServer(shutdownHandler shutdownHandler, handler http.Handler, addr Address, h2Only bool) (*Server, error) { listener, err := net.Listen(addr.Protocol(), addr.String()) if err != nil { return nil, err @@ -27,11 +28,26 @@ func StartServer(shutdownHandler shutdownHandler, handler http.Handler, addr Add addr.UpdateFromListener(listener) s := &Server{ - httpServer: &http.Server{Handler: handler, ReadHeaderTimeout: 15 * time.Second}, + httpServer: &http.Server{ + Handler: handler, + ReadHeaderTimeout: 15 * time.Second, + Protocols: &http.Protocols{}, + }, doneCh: make(chan struct{}), shutdownHandler: shutdownHandler, Addr: addr, } + if h2Only { + + s.httpServer.Protocols.SetUnencryptedHTTP2(true) + + s.httpServer.HTTP2 = &http.HTTP2Config{ + MaxConcurrentStreams: 2048, + } + } else { + + s.httpServer.Protocols.SetHTTP1(true) + } go func() { if err := s.httpServer.Serve(listener); err != nil { @@ -46,9 +62,17 @@ func (s *Server) Shutdown(err error) { s.shutdownOnce.Do(func() { s.shutdownHandler.Shutdown(model.NewClientError(err, model.ErrorSeverityFatal, model.ErrorExecutionEnvironmentShutdown)) slog.Info("Shutting down HTTP server...") - if err := s.httpServer.Close(); err != nil { - slog.Warn("error shutdown EA http server", "err", err) + + start := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) + defer cancel() + if err := s.httpServer.Shutdown(ctx); err != nil { + slog.Warn("EA server graceful shutdown timed out, forcing close", "err", err) + if closeErr := s.httpServer.Close(); closeErr != nil { + slog.Warn("EA server force close failed", "err", closeErr) + } } + slog.Info("EA HTTP server closed", "duration", time.Since(start)) if err != nil { s.err.Store(err) diff --git a/internal/lambda-managed-instances/raptor/server_test.go b/internal/lambda-managed-instances/raptor/server_test.go index eabae715..a574f85c 100644 --- a/internal/lambda-managed-instances/raptor/server_test.go +++ b/internal/lambda-managed-instances/raptor/server_test.go @@ -30,7 +30,7 @@ func TestStartNewServer_UDS(t *testing.T) { server, err := StartServer(mockShutdownHandler, handler, &UnixAddress{ Path: socketPath, - }) + }, true) require.NoError(t, err) assert.Equal(t, socketPath, server.Addr.String()) @@ -53,7 +53,7 @@ func TestStartNewServer_TCP(t *testing.T) { server, err := StartServer(mockShutdownHandler, handler, &TCPAddress{ eaAPIAddrPort, - }) + }, false) require.NoError(t, err) assert.Equal(t, eaAPIAddrPort, server.Addr.(*TCPAddress).AddrPort) @@ -69,7 +69,7 @@ func TestStartNewServer_UDS_ListenError(t *testing.T) { _, err := StartServer(mockShutdownHandler, handler, &UnixAddress{ Path: invalidSocketPath, - }) + }, true) assert.Error(t, err) assert.Contains(t, err.Error(), invalidSocketPath) } @@ -80,7 +80,7 @@ func TestStartNewServe_TCP_ListenError(t *testing.T) { mockShutdownHandler := newMockShutdownHandler(t) handler := mocks.NewNoOpHandler() - _, err := StartServer(mockShutdownHandler, handler, &TCPAddress{eaAPIAddrPort}) + _, err := StartServer(mockShutdownHandler, handler, &TCPAddress{eaAPIAddrPort}, false) assert.Error(t, err) assert.Contains(t, err.Error(), "1.1.1.1:49275") } diff --git a/internal/lambda-managed-instances/supervisor/local/process.go b/internal/lambda-managed-instances/supervisor/local/process.go index ec0bc0c3..9e64a501 100644 --- a/internal/lambda-managed-instances/supervisor/local/process.go +++ b/internal/lambda-managed-instances/supervisor/local/process.go @@ -11,6 +11,7 @@ import ( "os" "os/exec" "strconv" + "strings" "sync" "syscall" "time" @@ -159,7 +160,11 @@ func (s *ProcessSupervisor) Exec(ctx context.Context, req *model.ExecRequest) er cell = int32(status.Signal()) signo = &cell - cause = model.Signaled + if status.Signal() == syscall.SIGKILL && wasOomKill() { + cause = model.OomKilled + } else { + cause = model.Signaled + } } } } @@ -320,5 +325,43 @@ func (s *ProcessSupervisor) setProcessGroupPriorities(pid int) error { return fmt.Errorf("failed to set nice score for %d: %w", pgid, err) } + oomScorePath := fmt.Sprintf("/proc/%d/oom_score_adj", pid) + + f, err := os.OpenFile(oomScorePath, os.O_WRONLY, 0) + if err != nil { + return fmt.Errorf("could not open file %s: %w", oomScorePath, err) + } + defer func() { + if err := f.Close(); err != nil { + slog.Error("could not close file", "file", oomScorePath, "err", err) + } + }() + if _, err := f.WriteString("1000"); err != nil { + return fmt.Errorf("could not write to file %s: %w", oomScorePath, err) + } + return nil } + +func wasOomKill() bool { + return checkOomKill("/sys/fs/cgroup/memory.events") +} + +func checkOomKill(eventsPath string) bool { + data, err := os.ReadFile(eventsPath) + if err != nil { + slog.Warn("could not read memory.events", "path", eventsPath, "err", err) + return false + } + for _, line := range strings.Split(string(data), "\n") { + if after, found := strings.CutPrefix(line, "oom_kill "); found { + count, err := strconv.Atoi(after) + if err != nil { + slog.Warn("could not parse oom_kill count", "line", line, "err", err) + return false + } + return count > 0 + } + } + return false +} diff --git a/internal/lambda-managed-instances/supervisor/local/process_test.go b/internal/lambda-managed-instances/supervisor/local/process_test.go index bd1f10fb..fe5f83dc 100644 --- a/internal/lambda-managed-instances/supervisor/local/process_test.go +++ b/internal/lambda-managed-instances/supervisor/local/process_test.go @@ -6,6 +6,8 @@ package local import ( "context" "errors" + "os" + "path/filepath" "syscall" "testing" "time" @@ -226,3 +228,37 @@ func TestTerminateCheckStatus(t *testing.T) { require.NotNil(t, term.Signaled()) require.EqualValues(t, syscall.SIGTERM, *term.Signo) } + +func TestCheckOomKill_OomKilled(t *testing.T) { + path := filepath.Join(t.TempDir(), "memory.events") + os.WriteFile(path, []byte("low 0\nhigh 0\nmax 96\noom 1\noom_kill 1\noom_group_kill 0\n"), 0644) + assert.True(t, checkOomKill(path)) +} + +func TestCheckOomKill_NoOom(t *testing.T) { + path := filepath.Join(t.TempDir(), "memory.events") + os.WriteFile(path, []byte("low 0\nhigh 0\nmax 0\noom 0\noom_kill 0\noom_group_kill 0\n"), 0644) + assert.False(t, checkOomKill(path)) +} + +func TestCheckOomKill_FileNotFound(t *testing.T) { + assert.False(t, checkOomKill("/nonexistent/memory.events")) +} + +func TestCheckOomKill_MultipleOomKills(t *testing.T) { + path := filepath.Join(t.TempDir(), "memory.events") + os.WriteFile(path, []byte("low 0\nhigh 0\nmax 50\noom 3\noom_kill 3\noom_group_kill 0\n"), 0644) + assert.True(t, checkOomKill(path)) +} + +func TestCheckOomKill_MalformedCount(t *testing.T) { + path := filepath.Join(t.TempDir(), "memory.events") + os.WriteFile(path, []byte("low 0\nhigh 0\noom_kill abc\n"), 0644) + assert.False(t, checkOomKill(path)) +} + +func TestCheckOomKill_NoOomKillLine(t *testing.T) { + path := filepath.Join(t.TempDir(), "memory.events") + os.WriteFile(path, []byte("low 0\nhigh 0\nmax 0\n"), 0644) + assert.False(t, checkOomKill(path)) +} diff --git a/internal/lambda-managed-instances/testutils/blocking_read_closer.go b/internal/lambda-managed-instances/testutils/blocking_read_closer.go new file mode 100644 index 00000000..a9f5d5dc --- /dev/null +++ b/internal/lambda-managed-instances/testutils/blocking_read_closer.go @@ -0,0 +1,28 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package testutils + +import "io" + +type BlockingReadCloser struct { + done chan struct{} +} + +func NewBlockingReadCloser() *BlockingReadCloser { + return &BlockingReadCloser{done: make(chan struct{})} +} + +func (r *BlockingReadCloser) Read(p []byte) (int, error) { + <-r.done + return 0, io.EOF +} + +func (r *BlockingReadCloser) Close() error { + select { + case <-r.done: + default: + close(r.done) + } + return nil +} diff --git a/internal/lambda-managed-instances/testutils/functional/process_supervisor.go b/internal/lambda-managed-instances/testutils/functional/process_supervisor.go index 8fb4c22e..072470f8 100644 --- a/internal/lambda-managed-instances/testutils/functional/process_supervisor.go +++ b/internal/lambda-managed-instances/testutils/functional/process_supervisor.go @@ -40,9 +40,10 @@ type RuntimeEnv struct { } type RuntimeExecutionEnvironment struct { - Actions []ExecutionEnvironmentAction - InvokeID interop.InvokeID - RuntimeEnv *RuntimeEnv + Actions []ExecutionEnvironmentAction + InvokeID interop.InvokeID + InvocationID string + RuntimeEnv *RuntimeEnv } func (r *RuntimeEnv) Exec(request *model.ExecRequest) (<-chan struct{}, error) { @@ -134,6 +135,7 @@ func (r *RuntimeExecutionEnvironment) executeEnvActions(client *Client, t *testi if resp != nil { r.InvokeID = resp.Header.Get(invoke.RuntimeRequestIdHeader) + r.InvocationID = resp.Header.Get(invoke.RuntimeInvocationIdHeader) a.ValidateStatus(t, resp) } case StdoutAction: @@ -146,6 +148,9 @@ func (r *RuntimeExecutionEnvironment) executeEnvActions(client *Client, t *testi if a.InvokeID == "" { a.InvokeID = r.InvokeID } + if a.ResponseHeaders == nil && r.InvocationID != "" { + a.ResponseHeaders = map[string]string{invoke.RuntimeInvocationIdHeader: r.InvocationID} + } executeAndValidateAction(a, client, t) case InvocationStreamingResponseAction: if a.InvokeID == "" { @@ -156,6 +161,9 @@ func (r *RuntimeExecutionEnvironment) executeEnvActions(client *Client, t *testi if a.InvokeID == "" { a.InvokeID = r.InvokeID } + if a.ResponseHeaders == nil && r.InvocationID != "" { + a.ResponseHeaders = map[string]string{invoke.RuntimeInvocationIdHeader: r.InvocationID} + } executeAndValidateAction(a, client, t) case ExitAction: a.exitProcessOnce = r.RuntimeEnv.ExitProcessOnce diff --git a/internal/lambda-managed-instances/testutils/functional/runtime_actions.go b/internal/lambda-managed-instances/testutils/functional/runtime_actions.go index ea1559ae..503fcc92 100644 --- a/internal/lambda-managed-instances/testutils/functional/runtime_actions.go +++ b/internal/lambda-managed-instances/testutils/functional/runtime_actions.go @@ -157,13 +157,14 @@ type InvocationResponseAction struct { ContentType string InvokeID interop.InvokeID ResponseModeHeader string + ResponseHeaders map[string]string ExpectedStatus int ExpectedBody string Trailers map[string]string } func (a InvocationResponseAction) Execute(t *testing.T, client *Client) (*http.Response, error) { - return client.Response(a.InvokeID, a.Payload, a.ContentType, a.ResponseModeHeader, a.Trailers) + return client.ResponseWithHeaders(a.InvokeID, a.Payload, a.ContentType, a.ResponseModeHeader, a.Trailers, a.ResponseHeaders) } func (a InvocationResponseAction) ValidateStatus(t *testing.T, resp *http.Response) { @@ -185,17 +186,28 @@ func (a InvocationResponseAction) String() string { } type InvocationResponseErrorAction struct { - Payload string - ContentType string - InvokeID interop.InvokeID - ErrorType string - ErrorCause string - ExpectedStatus int - ExpectedBody string + Payload string + ContentType string + InvokeID interop.InvokeID + ErrorType string + ErrorCause string + ResponseHeaders map[string]string + ExpectedStatus int + ExpectedBody string } func (a InvocationResponseErrorAction) Execute(t *testing.T, client *Client) (*http.Response, error) { - return client.ResponseError(a.InvokeID, a.Payload, a.ContentType, a.ErrorType, a.ErrorCause) + headers := map[string]string{ContentTypeHeader: a.ContentType} + if len(a.ErrorType) > 0 { + headers[LambdaErrorTypeHeader] = a.ErrorType + } + if len(a.ErrorCause) > 0 { + headers[LambdaXRayErrorCauseHeader] = a.ErrorCause + } + for k, v := range a.ResponseHeaders { + headers[k] = v + } + return client.ResponseErrorWithHeaders(a.InvokeID, a.Payload, headers) } func (a InvocationResponseErrorAction) ValidateStatus(t *testing.T, resp *http.Response) { @@ -215,6 +227,7 @@ func (a InvocationResponseErrorAction) String() string { type InvocationStreamingResponseAction struct { Chunks []string + Body io.Reader ContentType string InvokeID interop.InvokeID ResponseModeHeader string @@ -223,10 +236,14 @@ type InvocationStreamingResponseAction struct { } func (a InvocationStreamingResponseAction) Execute(t *testing.T, client *Client) (*http.Response, error) { + var body io.Reader + if a.Body != nil { + body = a.Body + } else { + body = NewChunkedReader(a.Chunks, a.ChunkDelay) + } - chunkedReader := NewChunkedReader(a.Chunks, a.ChunkDelay) - - return client.Response(a.InvokeID, chunkedReader, a.ContentType, a.ResponseModeHeader, a.Trailers) + return client.Response(a.InvokeID, body, a.ContentType, a.ResponseModeHeader, a.Trailers) } func (a InvocationStreamingResponseAction) ValidateStatus(t *testing.T, resp *http.Response) {} diff --git a/internal/lambda-managed-instances/testutils/functional/runtime_client.go b/internal/lambda-managed-instances/testutils/functional/runtime_client.go index c8660169..34014b5b 100644 --- a/internal/lambda-managed-instances/testutils/functional/runtime_client.go +++ b/internal/lambda-managed-instances/testutils/functional/runtime_client.go @@ -77,12 +77,19 @@ func (client *Client) Next(body io.Reader) *http.Response { } func (client *Client) Response(invokeID interop.InvokeID, payload io.Reader, contentType string, responseModeHeader string, trailers map[string]string) (*http.Response, error) { + return client.ResponseWithHeaders(invokeID, payload, contentType, responseModeHeader, trailers, nil) +} + +func (client *Client) ResponseWithHeaders(invokeID interop.InvokeID, payload io.Reader, contentType string, responseModeHeader string, trailers map[string]string, extraHeaders map[string]string) (*http.Response, error) { url := fmt.Sprintf("%s/runtime/invocation/%s/response", client.baseurl, invokeID) headers := map[string]string{ContentTypeHeader: contentType} if responseModeHeader != "" { headers[LambdaResponseModeHeader] = responseModeHeader } + for k, v := range extraHeaders { + headers[k] = v + } return client.postBufferedResponse(url, payload, headers, trailers) } diff --git a/internal/lambda-managed-instances/testutils/socket_utils.go b/internal/lambda-managed-instances/testutils/socket_utils.go index b5fa7be9..3e3b5588 100644 --- a/internal/lambda-managed-instances/testutils/socket_utils.go +++ b/internal/lambda-managed-instances/testutils/socket_utils.go @@ -11,7 +11,6 @@ import ( "os" "path/filepath" "testing" - "time" "github.com/google/uuid" ) @@ -38,10 +37,22 @@ func NewUnixSocketClient(socketPath string) *http.Client { } transport := &http.Transport{ - DialContext: dialer, - DisableCompression: true, - ResponseHeaderTimeout: 5 * time.Second, + DialContext: dialer, + DisableCompression: true, + Protocols: &http.Protocols{}, } + transport.Protocols.SetUnencryptedHTTP2(true) + + return &http.Client{ + Transport: transport, + } +} + +func NewH2CClient() *http.Client { + transport := &http.Transport{ + Protocols: &http.Protocols{}, + } + transport.Protocols.SetUnencryptedHTTP2(true) return &http.Client{ Transport: transport,