Elastic Logstash

9 min

You can integrate metering data into Amberflo using the open-source tool Logstash.

How It Works

The integration utilizes the Logstash S3 output plugin to write metering records to your Amberflo-provided S3 bucket.
You can use Logstash's powerful parsing language to extract metering data from your logs or repositories.

Data Sources

Logstash supports a wide variety of input plugins, allowing you to extract data from:

Files
Elasticsearch
JDBC
S3
MongoDB
And many other systems

This flexibility makes it easy to integrate metering into your existing infrastructure. https://www.elastic.co/guide/en/logstash/current/input-plugins.html

Example - Logstash configuration

The following is a sample Logstash configuration that reads log lines from a file and writes the relevant metering records to Amberflo’s S3 bucket.

Input Setup

Logstash will tail files located at: /Users/demoaccount/input/*

Sample File Content

log line 1 log line 2 myMeter 2 myCustomerId amberflo_meter log line 3

How It Works

Logstash continuously monitors the input folder for new files.
It filters log lines using the condition: contains amberflo_meter.
Matching lines are split by space (" ").
The split values are mapped as follows:
- meterApiName → first item
- meterValue → second item
- customerId → third item

The records are then transformed using mutate logic and forwarded to Amberflo's S3 bucket.

input { 
   file {  
      path => "/Users/demoaccount/input/*" 
    } 
} 

filter {
    if "amberflo_meter" in [message]{
    ruby { code => "event.set('time', ((event.get('@timestamp').to_f*1000).to_i).to_s)" } 
     mutate {
            split => { "message" => " " }
            add_field => { "meterApiName" => "%{[message][0]}" }
            add_field => { "meterValue" => "%{[message][1]}" }
            add_field => { "customerId" => "%{[message][2]}" }
            remove_field => ["host","message","@version","@timestamp"]
        } 
    }else{
    drop {}
    }

}

output {
   s3{
     access_key_id => "XXXXXX"             
     secret_access_key => "YYYYYYYYYYYYYY" 
     region => "us-west-2"                    
     bucket => "demo-amberflo"               
     size_file => 2048                        
     time_file => 5                          
     codec => "json"                        
     canned_acl => "bucket-owner-full-control"                
   }

}

The resulting file (meter record) will be:

{
	"meterApiName": "myMeter",
	"meterValue": "2",
	"customerId": "myCustomerId",
	"time": "1621619742810",
	"path": "/Users/demoaccount/input/sample.txt"
}

JDBC source

You can use the Logstash JDBC input plugin to extract metering data directly from your database or repository.

This is useful when your metering data is stored in structured tables.

The plugin allows you to run SQL queries to retrieve only the relevant data for metering. For more details, refer to the official documentation: https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html

input {
  jdbc {
    statement => "SELECT id, mycolumn1, mycolumn2 FROM my_table WHERE id > :sql_last_value"
    use_column_value => true
    tracking_column => "id"
    jdbc_driver_library => "mysql-connector-java-5.1.36-bin.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"
    jdbc_user => "mysql"
    schedule => "* * * * *"
    # ... other configuration bits
  }
}

Elasticsearch input

If your metering data is stored in Elasticsearch logs, you can use the Elasticsearch input plugin to extract it into Logstash.

Here’s a basic example configuration:

input {
elasticsearch {
        hosts => "search-myes-cluster.us-west-2.es.amazonaws.com:443"
        index => "cwl--aws-lambda-meter-definition-api-lambda-2021.07"
        query => '{ "query": {"bool": {"filter": [{"range": {"@timestamp": {"from": "now-1d/d", "to": "now/d", "include_lower": true, "include_upper": true, "format": "epoch_millis", "boost": 1 } } }, {"query_string": {"query": "*amberflo_meter*", "default_field": "@message", "fields": [], "type": "best_fields", "default_operator": "or", "max_determinized_states": 10000, "enable_position_increments": true, "fuzziness": "AUTO", "fuzzy_prefix_length": 0, "fuzzy_max_expansions": 50, "phrase_slop": 0, "escape": false, "auto_generate_synonyms_phrase_query": true, "fuzzy_transpositions": true, "boost": 1 } } ], "adjust_pure_negative": true, "boost": 1 } }, "aggregations": {} }'
        size => 500
        scroll => "5m"
        docinfo => true
        ssl => true
        user => "myuser"
        password => "mypwd"
        #schedule => "*/1 * * * *"
      }

}