Send data from AWS S3 to Axiom

To determine the best method to send data from different AWS services, see Send data from AWS to Axiom.

Prerequisites#

Create an Axiom account.
Create a dataset in Axiom where you send your data.
Create an API token in Axiom with permissions to ingest data to the dataset you have created.
Create an AWS account with permissions to create and manage S3 buckets, Lambda functions, and IAM roles. For more information, see the AWS documentation.

Package the requests module#

Before creating the Lambda function, package the requests module so it can be used in the function:

Create a new directory.
Install the requests module into the current directory using pip.
Zip the contents of the directory.
Add your Lambda function file to the zip file.

Create AWS Lambda function#

Create a Lambda function with Python runtime and upload the packaged zip file containing the requests module and your function code below:

import os
import json
import boto3
import requests
import csv
import io
import ndjson
import re
 
def lambda_handler(event, context):
    # Extract the bucket name and object key from the event
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
 
    try:
        # Fetch the log file from S3
        s3 = boto3.client('s3')
        obj = s3.get_object(Bucket=bucket, Key=key)
    except Exception as e:
        print(f"Error fetching from S3: {str(e)}")
        raise e
 
    # Read the log data from the S3 object
    log_data = obj['Body'].read().decode('utf-8')
 
    # Determine the file format and parse accordingly
    file_extension = os.path.splitext(key)[1].lower()
 
    if file_extension == '.csv':
        csv_data = csv.DictReader(io.StringIO(log_data))
        json_logs = list(csv_data)
    elif file_extension == '.txt':
        log_lines = log_data.strip().split("\n")
        json_logs = [{'message': line} for line in log_lines]
    elif file_extension == '.log':
        # IMPORTANT: Log files can be in various formats (JSON, XML, syslog, etc.)
        try:
            # First, try to parse as JSON (either one JSON object per line or a JSON array)
            if log_data.strip().startswith('[') and log_data.strip().endswith(']'):
                # Appears to be a JSON array
                json_logs = json.loads(log_data)
            else:
                # Try parsing as NDJSON (one JSON object per line)
                try:
                    json_logs = ndjson.loads(log_data)
                except:
                    # If not valid NDJSON, check if each line might be JSON
                    log_lines = log_data.strip().split("\n")
                    json_logs = []
                    for line in log_lines:
                        try:
                            # Try to parse each line as JSON
                            parsed_line = json.loads(line)
                            json_logs.append(parsed_line)
                        except:
                            # Create a dictionary and let json module handle the escaping
                            message_dict = {'message': line}
                            json_logs.append(message_dict)
        except:
            # If JSON parsing fails, default to treating as plain text
            log_lines = log_data.strip().split("\n")
            json_logs = [{'message': line} for line in log_lines]
            print("Warning: Log file format could not be determined. Treating as plain text.")
    elif file_extension == '.ndjson' or file_extension == '.jsonl':
        json_logs = ndjson.loads(log_data)
    else:
        print(f"Unsupported file format: {file_extension}")
        return
 
    # Prepare Axiom API request
    dataset_name = os.environ['DATASET_NAME']
    axiom_edge_domain = os.environ['AXIOM_DOMAIN']
    axiom_api_url = f"https://{axiom_edge_domain}/v1/ingest/{dataset_name}"
    api_token = os.environ['API_TOKEN']
    axiom_headers = {
        "Authorization": f"Bearer {api_token}",
        "Content-Type": "application/json"
    }
 
    try:
        response = requests.post(axiom_api_url, headers=axiom_headers, json=json_logs)
        if response.status_code != 200:
            print(f"Failed to send logs to Axiom: {response.text}")
        else:
            print(f"Successfully sent logs to Axiom. Response: {response.text}")
    except Exception as e:
        print(f"Error sending to Axiom: {str(e)}")
 
    print(f"Processed {len(json_logs)} log entries")

In the environment variables section of the Lambda function configuration, add the following environment variables:

Replace API_TOKEN with the Axiom API token you have generated. For added security, store the API token in an environment variable.

Replace DATASET_NAME with the name of the Axiom dataset where you send your data.

Replace AXIOM_DOMAIN with the base domain of your edge deployment. For more information, see Edge deployments.

This example uses Python for the Lambda function. To use another language, change the code above accordingly.

Understanding log format#

The .log extension doesn't guarantee any specific format. Log files might contain:

JSON (single object or array)
NDJSON/JSONL (one JSON object per line)
Syslog format
XML
Application-specific formats (Apache, Nginx, ELB, etc.)
Custom formats with quoted strings and special characters

The example code includes format detection for common formats, but you’ll need to customize this based on your specific log structure.

Example: Custom parser for structured logs#

For logs with a specific structure (like AWS ELB logs), you have to implement a custom parser. Here’s a simplified example:

import shlex
import re 
 
class Parser:
    def parse_line(self, line):
        try:
            line = re.sub(r"[\[\]]", "", line)
            data = shlex.split(line)
            result = {
                "protocol": data[0],
                "timestamp": data[1],
                "client_ip_port": data[2],
                # ...more fields...
            }
            return result
        except Exception as e:
            raise e

Configure S3 to trigger Lambda#

In the Amazon S3 console, select the bucket where your log files are stored. Go to the properties tab, find the event notifications section, and create an event notification. Select All object create events as the event type and choose the Lambda function you created earlier as the destination. For more information, see the AWS documentation.

Upload a test log file#

Ensure the log file you upload to the S3 bucket is in the correct format, such as JSON or newline-delimited JSON (NDJSON) or CSV. Here’s an example:

[
   {
     "_time":"2021-02-04T03:11:23.222Z",
     "data":{"key1":"value1","key2":"value2"}
   },
   {
     "data":{"key3":"value3"},
     "attributes":{"key4":"value4"}
   },
   {
     "tags": {
       "server": "aws",
       "source": "wordpress"
     }
   }
 ]

After uploading a test log file to your S3 bucket, the Lambda function automatically processes the log data and sends it to Axiom. In Axiom, go to the Stream tab and select the dataset you specified in the Lambda function. You now see the logs from your S3 bucket in Axiom.

Updated Sep 25, 2026