From b9b61e2c90e1849af00c5741608e8afd9eba52ff Mon Sep 17 00:00:00 2001 From: Petr Heinz Date: Fri, 16 Oct 2026 14:26:59 +0200 Subject: [PATCH 01/13] E2E workflow: the AWS Lambda docs article, Node.js layer and Python Follows https://betterstack.com/docs/logs/aws-lambda/ as written: the article's code blocks are read from the page, the layer is built with the article's pnpm commands (latest pnpm, pnpm 9 when that fails) and published with its publish-layer-version command, and one function per runtime and architecture it lists runs its logging snippet with its environment variables. Python functions are packaged from its pip lines and run its setup and logging snippets. Each line must be stored once, as logged, with dt inside the invocation, in order, and received within 60 s. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/e2e.yml | 567 ++++++++++++++++++++ .github/workflows/notify_linear_failure.yml | 105 ++++ README.md | 1 + 3 files changed, 673 insertions(+) create mode 100644 .github/workflows/e2e.yml create mode 100644 .github/workflows/notify_linear_failure.yml diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml new file mode 100644 index 0000000..ac93c53 --- /dev/null +++ b/.github/workflows/e2e.yml @@ -0,0 +1,567 @@ +name: E2E + +# Follows the AWS Lambda docs article exactly as written, for both of its languages, in a throwaway AWS account: +# - Node.js: builds this repository with the article's `pnpm install` + `pnpm run build`, publishes the layer with the +# article's `aws lambda publish-layer-version` command, and runs one function per runtime and architecture that +# command lists, with the article's environment variables and its logging snippet as the handler. +# - Python: installs the article's pip packages into the deployment package and runs a function whose module is the +# article's setup snippet and whose handler is the article's logging snippet. +# Each function gets its own source, is invoked once, and every line the snippet logs must be stored once, with the +# level and fields it was logged with, a dt inside the invocation, in the order it was logged, and received by Better +# Stack (ingest_time) within DELIVERY_LIMIT_SECONDS of the invocation returning. The read-back waits up to +# WAIT_SECONDS to tell how late lines arrive. The source token and ingesting host are filled in where the article shows +# its $SOURCE_TOKEN and $INGESTING_HOST placeholders, as the article does for a signed-in reader. The docs job fails +# when the article's code blocks no longer have the shape this test reads. +# The layer job runs the article's build with the latest pnpm, as a reader would. When that fails, the job fails and +# pnpm 9 (the lockfile's version) builds the layer instead, so the Lambda jobs still run. +# Pull requests and pushes to main build the layer from the checkout; the daily schedule builds main, which is what the +# article's `git clone` gets. Everything a job creates is named after NAME and deleted in its cleanup step, together +# with leftovers of earlier runs of the same runtime and architecture older than two hours. Pull requests from forks +# are skipped: they get no OIDC token. +# See the docs article in https://betterstack.com/docs/logs/aws-lambda/ +# +# AWS account: better-stack-e2e-lambda (352333568934), shared with logtail/logtail-aws-lambda's E2E. Root sign-in: +# 1Password, Engineering vault, item "AWS better-stack-e2e-lambda". This workflow assumes IAM role +# `github-actions-e2e-lambda-docs`, which trusts this repository through the account's GitHub OIDC provider and may +# only touch functions, layers and log groups named `e2e-lambda-docs-*`, so it cannot reach the other E2E's +# resources. The functions run as the standing IAM role `e2e-lambda-exec`. +# Secrets: TEST_TEAM_TELEMETRY_API, TEST_TEAM_CONNECTION_USERNAME, TEST_TEAM_CONNECTION_PASSWORD (the E2E testing team), +# LINEAR_* (see notify_linear_failure.yml). + +on: + push: + branches: [main] + pull_request: + schedule: + - cron: '45 5 * * *' + workflow_dispatch: + +permissions: + id-token: write + contents: read + actions: read + +env: + AWS_REGION: us-east-1 + AWS_ROLE_ARN: arn:aws:iam::352333568934:role/github-actions-e2e-lambda-docs + LAMBDA_ROLE_ARN: arn:aws:iam::352333568934:role/e2e-lambda-exec + API: https://telemetry.betterstack.com + DOCS_URL: https://betterstack.com/docs/logs/aws-lambda/ + RUN_ID: gh${{ github.run_id }}a${{ github.run_attempt }} + # The article asks for Python 3.7 or higher; these are the oldest and the newest Python runtimes Lambda supports. + PYTHON_RUNTIMES: python3.11 python3.14 + DELIVERY_LIMIT_SECONDS: 60 + WAIT_SECONDS: 900 + TEST_TEAM_TELEMETRY_API: ${{ secrets.TEST_TEAM_TELEMETRY_API }} + TEST_TEAM_CONNECTION_USERNAME: ${{ secrets.TEST_TEAM_CONNECTION_USERNAME }} + TEST_TEAM_CONNECTION_PASSWORD: ${{ secrets.TEST_TEAM_CONNECTION_PASSWORD }} + +jobs: + docs: + name: Docs article + if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }} + runs-on: ubuntu-latest + timeout-minutes: 5 + outputs: + matrix: ${{ steps.read.outputs.matrix }} + layer_zip: ${{ steps.read.outputs.layer_zip }} + + steps: + - name: "[Setup] Fetch the article" + run: curl -sSfL "$DOCS_URL" -o docs.html + + - name: "[Check] Read the article's code blocks" + id: read + shell: python + run: | + import ast, html, json, os, re, shlex, sys, textwrap + + repository = os.environ['GITHUB_REPOSITORY'] + page = open('docs.html', encoding='utf-8').read() + blocks = [(language, html.unescape(re.sub(r'<[^>]+>', '', code)).strip()) + for language, code in re.findall(r"
(.*?)
", page, re.S)] + ROLES = ['Node.js build', 'Node.js environment variables', 'Node.js logging', 'Python install', 'Python setup', 'Python logging'] + LANGUAGES = ['bash', 'text', 'js', 'bash', 'python', 'python'] + if [language for language, _ in blocks] != LANGUAGES: + sys.exit(f"::error::{os.environ['DOCS_URL']} changed: its code blocks are now {[language for language, _ in blocks]}, " + f"this test reads {LANGUAGES} as {ROLES}") + node_build, node_env, node_logging, python_install, python_setup, python_logging = [code for _, code in blocks] + PLACEHOLDERS = {'$SOURCE_TOKEN', '$INGESTING_HOST'} + problems = [] + + # Node.js build: clone this repository, build it, publish the ZIP as a layer. + commands = [line.strip() for line in node_build.replace('\\\n', ' ').splitlines() if line.strip()] + steps = [f'git clone git@github.com:{repository}.git', f'cd {repository.split("/")[1]}', 'pnpm install', 'pnpm run build'] + if commands[:-1] != steps: + problems.append(f'Node.js build commands are {commands[:-1]}, this test runs {steps}') + publish = shlex.split(commands[-1]) if commands else [] + options, option = {}, None + for word in publish[3:]: + if word.startswith('--'): + option = word + options[option] = [] + elif option: + options[option].append(word) + else: + options.setdefault(None, []).append(word) + if (publish[:3] != ['aws', 'lambda', 'publish-layer-version'] + or sorted(options, key=str) != ['--compatible-architectures', '--compatible-runtimes', '--layer-name', '--zip-file'] + or len(options['--zip-file']) != 1 or not options['--zip-file'][0].startswith('fileb://') + or len(options['--layer-name']) != 1 + or not all(re.fullmatch(r'nodejs\d+\.x', runtime) for runtime in options['--compatible-runtimes']) + or not set(options['--compatible-architectures']) <= {'x86_64', 'arm64'}): + problems.append(f'the layer publish command is {commands[-1:]}, this test reads --layer-name, --zip-file fileb://…, ' + f'--compatible-runtimes nodejs.x … and --compatible-architectures x86_64/arm64') + options = {'--zip-file': ['fileb://'], '--compatible-runtimes': [], '--compatible-architectures': []} + + # Node.js environment variables: "NAME: value" lines, values may carry the article's placeholders. + environment = {} + for line in node_env.splitlines(): + match = re.fullmatch(r'\s*([A-Z][A-Z0-9_]*):\s*(\S+)\s*', line) + if not match: + problems.append(f'Node.js environment variable line {line!r} is not "NAME: value"') + continue + environment[match[1]] = match[2] + unknown = {placeholder for value in environment.values() for placeholder in re.findall(r'\$[A-Z_]+', value)} - PLACEHOLDERS + if unknown: + problems.append(f'Node.js environment variables use placeholders this test cannot fill: {sorted(unknown)}') + + # Node.js logging: console.('message'[, {fields}]) calls, in the Lambda text format's level names. + LAMBDA_LEVELS = {'log': 'INFO', 'info': 'INFO', 'warn': 'WARN', 'error': 'ERROR', 'debug': 'DEBUG', 'trace': 'TRACE'} + CALL = re.compile(r"console\.(\w+)\(\s*'([^'\\]*)'\s*(?:,\s*(\{[^{}]*\})\s*)?\)\s*;?", re.S) + node_lines = [] + for call in CALL.finditer(node_logging): + try: + fields = json.loads(call[3].replace("'", '"')) if call[3] else {} + except ValueError: + problems.append(f'cannot read the fields of {call[0]!r}') + fields = {} + if call[1] not in LAMBDA_LEVELS: + problems.append(f'console.{call[1]} has no known Lambda log level') + node_lines.append({'level': LAMBDA_LEVELS.get(call[1]), 'message': call[2], 'fields': fields}) + if CALL.sub('', node_logging).strip() or not node_lines: + problems.append(f'the Node.js logging snippet holds code this test cannot read: {CALL.sub("", node_logging).strip()!r}') + + # Python install: "pip3 install " lines. + packages = [] + for line in python_install.splitlines(): + words = line.split() + if len(words) == 3 and words[:2] == ['pip3', 'install']: + packages.append(words[2]) + else: + problems.append(f'Python install line {line!r} is not "pip3 install "') + + # Python setup: module code with both placeholders. + try: + ast.parse(python_setup) + except SyntaxError as error: + problems.append(f'the Python setup snippet does not parse: {error}') + if set(re.findall(r'\$[A-Z_]+', python_setup)) != PLACEHOLDERS: + problems.append(f'the Python setup snippet has placeholders {sorted(set(re.findall(r"[$][A-Z_]+", python_setup)))}, expected {sorted(PLACEHOLDERS)}') + + # Python logging: logger.('message'[, extra={fields}]) calls. + python_lines = [] + try: + statements = ast.parse(python_logging).body + except SyntaxError as error: + problems.append(f'the Python logging snippet does not parse: {error}') + statements = [] + for statement in statements: + call = statement.value if isinstance(statement, ast.Expr) else None + if not (isinstance(call, ast.Call) and isinstance(call.func, ast.Attribute) and isinstance(call.func.value, ast.Name) + and call.func.value.id == 'logger' and len(call.args) == 1 and isinstance(call.args[0], ast.Constant) + and all(keyword.arg == 'extra' for keyword in call.keywords)): + problems.append(f'the Python logging snippet holds code this test cannot read: {ast.unparse(statement)!r}') + continue + fields = ast.literal_eval(call.keywords[0].value) if call.keywords else {} + python_lines.append({'level': call.func.attr, 'message': call.args[0].value, 'fields': fields}) + if not python_lines: + problems.append('the Python logging snippet logs nothing') + + if problems: + print('\n'.join(f"::error::{os.environ['DOCS_URL']} changed, this test no longer matches it: {problem}" for problem in problems)) + sys.exit(1) + + os.makedirs('article', exist_ok=True) + # The article shows the logging snippets without a handler around them; they become the handler's body. + with open('article/index.mjs', 'w') as file: + file.write('export const handler = async () => {\n' + textwrap.indent(node_logging, ' ') + '\n};\n') + with open('article/lambda_function.py', 'w') as file: + file.write(python_setup + '\n\n\ndef lambda_handler(event, context):\n' + textwrap.indent(python_logging, ' ') + '\n') + with open('article/article.json', 'w') as file: + json.dump({ + 'layer': {'runtimes': options['--compatible-runtimes'], 'architectures': options['--compatible-architectures']}, + 'environment': environment, + 'packages': packages, + 'lines': {'node': node_lines, 'python': python_lines}, + }, file, indent=2) + + include = [{'language': 'node', 'runtime': runtime, 'arch': arch, 'variant': f"{runtime.replace('.', '')}-{arch}"} + for runtime in options['--compatible-runtimes'] for arch in options['--compatible-architectures']] + include += [{'language': 'python', 'runtime': runtime, 'arch': 'x86_64', 'variant': f"{runtime.replace('.', '')}-x86_64", + 'python': runtime.removeprefix('python')} for runtime in os.environ['PYTHON_RUNTIMES'].split()] + with open(os.environ['GITHUB_OUTPUT'], 'a') as output: + output.write(f"matrix={json.dumps({'include': include})}\n") + output.write(f"layer_zip={options['--zip-file'][0].removeprefix('fileb://')}\n") + with open(os.environ['GITHUB_STEP_SUMMARY'], 'a') as summary: + summary.write(f"### Docs article\n\n{os.environ['DOCS_URL']} matches what this test runs:\n\n" + f"- layer for {', '.join(options['--compatible-runtimes'])} on {', '.join(options['--compatible-architectures'])}\n" + f"- Node.js environment variables {', '.join(f'`{name}: {value}`' for name, value in environment.items())}\n" + f"- Python packages {', '.join(f'`{package}`' for package in packages)}, tested on {os.environ['PYTHON_RUNTIMES']}\n" + f"- {len(node_lines)} Node.js and {len(python_lines)} Python log lines\n") + print(json.dumps(include, indent=2)) + + - name: "[Setup] Share the article's snippets with the Lambda jobs" + uses: actions/upload-artifact@v7 + with: + name: article + path: article/ + if-no-files-found: error + + layer: + name: Layer build as documented + needs: docs + runs-on: ubuntu-latest + timeout-minutes: 15 + outputs: + built_with: ${{ steps.report.outputs.built_with }} + + steps: + - name: "[Setup] Checkout code (the article's git clone)" + uses: actions/checkout@v7 + + - name: "[Setup] Node.js 22, the article's minimum" + uses: actions/setup-node@v7 + with: + node-version: 22.x + + # Outside pnpm's CI mode, as in a reader's terminal. + - name: "[Test] pnpm install and pnpm run build with the latest pnpm" + id: latest + continue-on-error: true + run: | + npm install --global pnpm@latest + echo "LATEST_PNPM=$(pnpm --version)" >> "$GITHUB_ENV" + env -u CI pnpm install + env -u CI pnpm run build + + - name: "[Setup] Build with pnpm 9 instead, so the Lambda jobs still run" + if: ${{ steps.latest.outcome == 'failure' }} + run: | + git checkout -- . + git clean -fdxq + npm install --global pnpm@9 + env -u CI pnpm install + env -u CI pnpm run build + + - name: "[Setup] Share the layer ZIP with the Lambda jobs" + uses: actions/upload-artifact@v7 + with: + name: layer + path: ${{ needs.docs.outputs.layer_zip }} + if-no-files-found: error + + - name: "[Check] The article's build works with the latest pnpm" + id: report + if: ${{ always() }} + run: | + BUILT_WITH="pnpm $(pnpm --version)" + echo "built_with=$BUILT_WITH" >> "$GITHUB_OUTPUT" + { + echo "### Layer build" + echo + if [ "${{ steps.latest.outcome }}" = success ]; then + echo "✅ \`pnpm install\` + \`pnpm run build\` with the latest pnpm ($LATEST_PNPM) on Node.js $(node --version)" + else + echo "❌ \`pnpm install\` + \`pnpm run build\` failed with the latest pnpm (${LATEST_PNPM:-unknown}) on Node.js $(node --version), see the step log; the Lambda jobs use the layer built with $BUILT_WITH" + fi + } >> "$GITHUB_STEP_SUMMARY" + [ "${{ steps.latest.outcome }}" = success ] || { echo "::error::the article's pnpm install + pnpm run build failed with the latest pnpm (${LATEST_PNPM:-unknown})"; exit 1; } + + lambda: + name: ${{ matrix.runtime }} ${{ matrix.arch }} + needs: [docs, layer] + # Runs when only the latest-pnpm build failed: the layer job then shares the pnpm 9 build. + if: ${{ !cancelled() && needs.docs.result == 'success' }} + runs-on: ubuntu-latest + timeout-minutes: 45 + strategy: + fail-fast: false + matrix: ${{ fromJSON(needs.docs.outputs.matrix) }} + env: + LANGUAGE: ${{ matrix.language }} + RUNTIME: ${{ matrix.runtime }} + ARCH: ${{ matrix.arch }} + NAME: e2e-lambda-docs-${{ matrix.variant }}-gh${{ github.run_id }}a${{ github.run_attempt }} + NAME_PREFIX: e2e-lambda-docs-${{ matrix.variant }}-gh + + steps: + - name: "[Setup] The article's snippets" + uses: actions/download-artifact@v8 + with: + name: article + path: article + + - name: "[Setup] The layer ZIP" + if: ${{ matrix.language == 'node' }} + uses: actions/download-artifact@v8 + with: + name: layer + path: layer + + - name: "[Setup] Python ${{ matrix.python }}, matching the runtime" + if: ${{ matrix.language == 'python' }} + uses: actions/setup-python@v7 + with: + python-version: ${{ matrix.python }} + + - name: "[Setup] AWS credentials (OIDC)" + uses: aws-actions/configure-aws-credentials@v6 + with: + role-to-assume: ${{ env.AWS_ROLE_ARN }} + aws-region: ${{ env.AWS_REGION }} + + - name: "[Setup] Create the source" + run: | + PLATFORM=${{ matrix.language == 'node' && 'javascript' || 'python' }} + RES=$(curl -sS -X POST "$API/api/v1/sources" -H "Authorization: Bearer $TEST_TEAM_TELEMETRY_API" -H 'Content-Type: application/json' \ + -d "$(jq -nc --arg name "$NAME" --arg platform "$PLATFORM" '{name: $name, platform: $platform}')") + jq -e '.data.id' <<<"$RES" >/dev/null || { echo "::error::creating the source failed: $RES"; exit 1; } + SOURCE_TOKEN=$(jq -r '.data.attributes.token' <<<"$RES") + echo "::add-mask::$SOURCE_TOKEN" + { + echo "SOURCE_ID=$(jq -r '.data.id' <<<"$RES")" + echo "SOURCE_TOKEN=$SOURCE_TOKEN" + echo "INGESTING_HOST=$(jq -r '.data.attributes.ingesting_host' <<<"$RES")" + echo "COLLECTION=t$(jq -r '.data.attributes.team_id' <<<"$RES")_$(jq -r '.data.attributes.table_name' <<<"$RES")" + echo "CONNECT_HOST=$(jq -r '.data.attributes.data_region' <<<"$RES")-connect.betterstackdata.com" + } >> "$GITHUB_ENV" + echo "$PLATFORM source $(jq -r '.data.id' <<<"$RES") on $(jq -r '.data.attributes.data_region' <<<"$RES")" + + # A fresh token reaches the ingesters with a delay; until then they answer 401. Send warm-up lines until 20 in a + # row are accepted, then let it settle, so a late line is never the token's fault. + - name: "[Setup] Warm up the source until ingest accepts its token" + run: | + STREAK=0 + DEADLINE=$((SECONDS + 300)) + while [ "$STREAK" -lt 20 ]; do + CODE=$(curl -sS -o /dev/null -w '%{http_code}' -X POST "https://$INGESTING_HOST/" -H "Authorization: Bearer $SOURCE_TOKEN" -H 'Content-Type: application/json' -d "{\"message\":\"e2e warm-up run_id=$RUN_ID\"}") + if [ "$CODE" = 202 ]; then STREAK=$((STREAK + 1)); else echo "ingest answered HTTP $CODE after $STREAK accepted"; STREAK=0; fi + [ "$SECONDS" -lt "$DEADLINE" ] || { echo "::error::ingest did not accept the source token 20 times in a row within 5 min"; exit 1; } + sleep 1 + done + sleep 15 + echo "ingest accepts the token" + + - name: "[Test] Publish the layer with the article's command" + if: ${{ matrix.language == 'node' }} + run: | + LAYER_ARN=$(aws lambda publish-layer-version --layer-name "$NAME" --zip-file "fileb://layer/$(basename '${{ needs.docs.outputs.layer_zip }}')" \ + --compatible-runtimes $(jq -r '.layer.runtimes | join(" ")' article/article.json) \ + --compatible-architectures $(jq -r '.layer.architectures | join(" ")' article/article.json) \ + --query LayerVersionArn --output text) + echo "LAYER_ARN=$LAYER_ARN" >> "$GITHUB_ENV" + echo "published $LAYER_ARN" + + - name: "[Test] Package the function from the article's snippets" + run: | + mkdir package + if [ "$LANGUAGE" = node ]; then + cp article/index.mjs package/index.mjs + echo "PACKAGE_VERSIONS=layer built with ${{ needs.layer.outputs.built_with }}" >> "$GITHUB_ENV" + else + pip3 install --quiet --target package $(jq -r '.packages | join(" ")' article/article.json) + # The placeholders in the article's setup snippet are filled in as the article does for a signed-in reader. + CODE=$(cat article/lambda_function.py) + CODE=${CODE//'$SOURCE_TOKEN'/$SOURCE_TOKEN} + CODE=${CODE//'$INGESTING_HOST'/$INGESTING_HOST} + printf '%s\n' "$CODE" > package/lambda_function.py + echo "PACKAGE_VERSIONS=$(cd package && ls -d *.dist-info | sed -E 's/\.dist-info$//' | grep -iE "^($(jq -r '.packages | map(gsub("-"; "_")) | join("|")' article/article.json))-" | tr '\n' ' ')" >> "$GITHUB_ENV" + fi + (cd package && zip -qr ../function.zip .) + echo "$(unzip -l function.zip | tail -1)" + + - name: "[Test] Create the function with the article's settings" + run: | + if [ "$LANGUAGE" = node ]; then + # The article's environment variables, with its placeholders filled in. + ENVIRONMENT=$(jq -c --arg token "$SOURCE_TOKEN" --arg host "$INGESTING_HOST" '{Variables: (.environment | map_values(gsub("\\$SOURCE_TOKEN"; $token) | gsub("\\$INGESTING_HOST"; $host)))}' article/article.json) + SETTINGS=(--handler index.handler --layers "$LAYER_ARN" --environment "$ENVIRONMENT") + else + SETTINGS=(--handler lambda_function.lambda_handler) + fi + # The response would echo the environment variables, token included; nothing of it is needed. + aws lambda create-function --function-name "$NAME" --runtime "$RUNTIME" --architectures "$ARCH" --role "$LAMBDA_ROLE_ARN" \ + --zip-file fileb://function.zip "${SETTINGS[@]}" >/dev/null + echo "FUNCTION_CREATED=true" >> "$GITHUB_ENV" + aws lambda wait function-active-v2 --function-name "$NAME" + echo "created $NAME on $RUNTIME $ARCH" + + - name: "[Test] Invoke the function once" + run: | + INVOKED_AT_MS=$(date +%s%3N) + aws lambda invoke --function-name "$NAME" --log-type Tail --cli-read-timeout 120 response.json > invoke.json + RETURNED_AT_MS=$(date +%s%3N) + echo "INVOKED_AT_MS=$INVOKED_AT_MS" >> "$GITHUB_ENV" + echo "RETURNED_AT_MS=$RETURNED_AT_MS" >> "$GITHUB_ENV" + echo "invocation took $((RETURNED_AT_MS - INVOKED_AT_MS)) ms, response $(cat response.json), its log tail:" + jq -r '.LogResult' invoke.json | base64 -d + jq -e '.StatusCode == 200 and .FunctionError == null' invoke.json >/dev/null || { echo "::error::the invocation failed: $(jq -c 'del(.LogResult)' invoke.json)"; exit 1; } + + - name: "[Test] Read the lines back through the SQL API connection" + shell: python + run: | + import base64, json, os, sys, time, urllib.error, urllib.request + + env = os.environ + language = env['LANGUAGE'] + expected = json.load(open('article/article.json'))['lines'][language] + invoked_at, returned_at = int(env['INVOKED_AT_MS']), int(env['RETURNED_AT_MS']) + limit = int(env['DELIVERY_LIMIT_SECONDS']) + credentials = base64.b64encode(f"{env['TEST_TEAM_CONNECTION_USERNAME']}:{env['TEST_TEAM_CONNECTION_PASSWORD']}".encode()).decode() + + def sql(query): + request = urllib.request.Request(f"https://{env['CONNECT_HOST']}/", data=f'{query} FORMAT JSONEachRow'.encode(), + headers={'Authorization': f'Basic {credentials}', 'Content-Type': 'plain/text'}) + try: + with urllib.request.urlopen(request, timeout=60) as response: + return [json.loads(line) for line in response.read().decode().splitlines() if line] + except urllib.error.HTTPError as error: + body = error.read().decode() + # A source's collection exists only from its first stored row on; until then the query is "no rows yet". + if error.code == 500 and 'CLUSTER_DOESNT_EXIST' in body: + return [] + sys.exit(f"::error::SQL API on {env['CONNECT_HOST']} answered HTTP {error.code}: {body[:300]}") + + # The collection belongs to this job's source alone, so apart from the warm-up lines every row is the function's. + QUERY = (f"SELECT toUnixTimestamp64Milli(dt) AS dt_ms, toUnixTimestamp64Milli(ingest_time) AS ingest_ms, raw " + f"FROM remote({env['COLLECTION']}_logs) WHERE position(raw, 'e2e warm-up run_id=') = 0 ORDER BY dt, _insert_index") + deadline = time.time() + int(env['WAIT_SECONDS']) + while True: + rows = sql(QUERY) + print(f"{len(rows)}/{len(expected)} rows, {time.time() - returned_at / 1000:.0f} s after the invocation returned", flush=True) + if len(rows) >= len(expected) or time.time() > deadline: + break + time.sleep(5) + if rows: + # A last look 30 s later catches duplicate delivery. + time.sleep(30) + rows = sql(QUERY) + for row in rows: + row['dt_ms'], row['ingest_ms'], row['raw'] = int(row['dt_ms']), int(row['ingest_ms']), json.loads(row['raw']) + + problems = [] + if len(rows) != len(expected): + problems.append(f"{len(rows)} rows stored within {env['WAIT_SECONDS']} s, the article's snippet logs {len(expected)} lines") + table, matched = [], [] + for n, line in enumerate(expected, 1): + hits = [index for index, row in enumerate(rows) if line['message'] in str(row['raw'].get('message', ''))] + if len(hits) != 1: + if hits or rows: + problems.append(f"line {n} ({line['message']!r}) is in {len(hits)} rows, expected 1") + table.append(f"| {n} | `{line['level']}` {line['message']} | ❌ {len(hits)} rows | | |") + continue + row = rows[hits[0]] + matched.append(hits[0]) + message = str(row['raw']['message']) + if language == 'node': + # Lambda's text log format: "