Elastic Logstash

5min
Meter Ingestion via Logstash
Integrating metering data into Amberflo can be done with open source Logstash.
The integration uses the Logstash S3 output plugin and writes the metering records to the Amberflo S3 bucket.
You can use Logstash's rich language for parsing the metering data out of your logs/repositories.
Logstash can extract data from files, Elasticsearch, JDBC, S3, MongoDB and other systems using various input plugins. https://www.elastic.co/guide/en/logstash/current/input-plugins.html﻿
Example - Logstash configuration
In the example below we will share a sample Logstash configuration that reads log lines from a file and writes the relevant metering records to Amberflo's S3 bucket.
We will tail new files locally from /Users/demoaccount/input/*
the sample file format
log line 1
log line 2
myMeter 2 myCustomerId amberflo_meter
log line 3
Logstash will read any new file and transform them using the mutate logic into metering records.
In this example we ignore any line without the string: amberflo_meter. We splits based on the space character " ", and treat the meterApiName, meterValue, cusomterId as the first, second and third object accordingly.
Ruby
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"                
   }

}
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:
JSON
{
	"meterApiName": "myMeter",
	"meterValue": "2",
	"customerId": "myCustomerId",
	"time": "1621619742810",
	"path": "/Users/demoaccount/input/sample.txt"
}
{
	"meterApiName": "myMeter",
	"meterValue": "2",
	"customerId": "myCustomerId",
	"time": "1621619742810",
	"path": "/Users/demoaccount/input/sample.txt"
}
﻿
JDBC source
You can use the Logstash JDBC source to extract the metering info from your repository.
https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html﻿
Ruby
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
  }
}
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
You can use the Elasticsearch input if you have metering data in your logs.
An example input plugin looks like:
Ruby
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 * * * *"
      }

}
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 * * * *"
      }

}
﻿
﻿
Updated 05 Nov 2024
Did this page help you?
AWS SQS
Databases and JDBC