Send data from AWS S3 to Axiom

Efficiently send log data from AWS S3 to Axiom via Lambda function

This page explains how to set up an AWS Lambda function to send logs from an S3 bucket to Axiom. The Lambda function triggers when a new log file is uploaded to an S3 bucket, processes the log data, and sends it 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:

python

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.

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:

python

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:

JSON

[
   {
     "_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.