---
title: Elastic Logstash
slug: 6wj7zaYyzNJpMCJV0_G-f
docTags: 
createdAt: 2024-07-11T15:15:35.977Z
---

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](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

:::BlockQuote
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**.

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

}
```

The resulting file (meter record) will be:

```json
{
	"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](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
  }
}
```

## 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:

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

}
```