input { 
  s3 {
    # KEY / Secret de amazon, editar segun convenga
    access_key_id => "AKIATITELAVOYADECIR"
    secret_access_key => "876af98asd6f98ew6fasd98f"
    # Bucket de s3 dodne se guarda el billing
    bucket => "billing-bucket"
    # prefijo (comienzo del nombre de fichero)
    prefix => "555555555555555-aws-billing-csv-"
  }
}

filter {
  csv {
    ## Columnas extraidas tal cual de la primera linea de la factura

    columns => ["InvoiceID","PayerAccountId","LinkedAccountId","RecordType",
       "RecordID","BillingPeriodStartDate","BillingPeriodEndDate",
       "InvoiceDate","PayerAccountName","LinkedAccountName","TaxationAddress",
       "PayerPONumber","ProductCode","ProductName","SellerOfRecord","UsageType",
       "Operation","RateId","ItemDescription","UsageStartDate","UsageEndDate",
       "UsageQuantity","BlendedRate","CurrencyCode","CostBeforeTax","Credits",
       "TaxAmount","TaxType","TotalCost"]

    ## Columnas que seran indexadas como tipo "float" (si no lo son es imposible hacer cálculos)

    convert => { "UsageQuantity" => "float" }
    convert => { "BlendedRate" => "float" }
    convert => { "CostBeforeTax" => "float" }
    convert => { "Credits" => "float" }
    convert => { "TaxAmount" => "float" }
    convert => { "TotalCost" => "float" }
  }
  # Retirar todo lo que no sean cargos
  if [RecordType] not in ["LinkedLineItem","PayerLineItem"] { drop {} }

  ## Fecha del evento modificada
  date {
    locale => "en_US"
    match => ["BillingPeriodEndDate","yyyy/MM/dd HH:mm:ss"]
    tag_on_failure => ["date_error"]
  }

  ## Campos de fecha que queramos guardar como tal
  date {
    target => "BillingPeriodStartDate"
    locale => "en_US"
    match => ["BillingPeriodStartDate","yyyy/MM/dd HH:mm:ss"]
    tag_on_failure => ["date_error"]
  }
  date {
    target => "UsageStartDate"
    locale => "en_US"
    match => ["UsageStartDate","yyyy/MM/dd HH:mm:ss"]
    tag_on_failure => ["date_error"]
  }
  date {
    target => "UsageEndDate"
    locale => "en_US"
    match => ["UsageEndDate","yyyy/MM/dd HH:mm:ss"]
    tag_on_failure => ["date_error"]
  }
  date {
    target => "BillingPeriodEndDate"
    locale => "en_US"
    match => ["BillingPeriodEndDate","yyyy/MM/dd HH:mm:ss"]
    tag_on_failure => ["date_error"]
  }
}


## Salida
output {
  if [date_error] in [tags] {
    ## mostrar fallidos si hay error de conversion de fecha
    stdout { codec => plain }
  } else {
    ## Volcar como un punto en pantalla y en elastic
    stdout { codec => dots }
    elasticsearch {
      hosts => ["elasticsearch:9200"]
    }
  }
}