logstash pipline output

fendys.myqnapcloud.com kafka -> local server file

input {
        kafka {
                bootstrap_servers => "fendys.myqnapcloud.com:9094"
                topics_pattern => ".*"
                codec => json
                group_id => "all-log-idc-backup"
                consumer_threads => "10"
        }
}

filter {
    if [service][name] {
      if [service][type] {
        mutate { replace => { "[@metadata][archive][namespace]" => "%{[service][name]}/%{+YYYY}-%{+MM}-%{+dd}-%{[service][name]}-%{[service][type]}.log" } }
      } else {
        mutate { replace => { "[@metadata][archive][namespace]" => "%{[service][name]}/%{+YYYY}-%{+MM}-%{+dd}-none_type.log" } }
      }
    }
    else if [cloud][service][name] {
      mutate { replace => { "[@metadata][archive][namespace]" => "%{[cloud][service][name]}/%{+YYYY}-%{+MM}-%{+dd}-%{[cloud][service][name]}.log" } }
    }
    else {
        if [service][type] {
      mutate { replace => { "[@metadata][archive][namespace]" => "none_name/%{+YYYY}-%{+MM}-%{+dd}-%{[service][type]}.log" } }
    } else {
      mutate { replace => { "[@metadata][archive][namespace]" => "none_name/%{+YYYY}-%{+MM}-%{+dd}-none_type.log" } }
    }
}

    if ![@metadata][archive][namespace] {
      mutate { replace => { "[@metadata][archive][namespace]" => "%{[@metadata][kafka][topic]}/%{+YYYY}-%{+MM}-%{+dd}-%{[@metadata][kafka][topic]}.log" } }
    }
}

output {
        file {
                path => "/data/%{+YYYY}/%{+MM}/%{+dd}/%{[@metadata][archive][namespace]}"
                codec => "json_lines"
}
}

 

fendys.myqnapcloud.com kafka -> aws s3 bucket

input {
  kafka {
    bootstrap_servers => "fendys.myqnapcloud.com:9094"
    topics_pattern => ".*"
    group_id => "log-archiver"
    codec => json
    decorate_events => true
  }
}

filter {
    if [service][name] {
      if [service][type] {
        mutate { replace => { "[@metadata][archive][namespace]" => "%{[service][name]}/%{+YYYY}/%{+MM}-%{+dd}/%{[service][type]}" } }
      } else {
        mutate { replace => { "[@metadata][archive][namespace]" => "%{[service][name]}/%{+YYYY}/%{+MM}-%{+dd}/none_type" } }
      }
    }
    else if [cloud][service][name] {
      mutate { replace => { "[@metadata][archive][namespace]" => "%{[cloud][service][name]}/%{+YYYY}/%{+MM}-%{+dd}" } }
    }
    else {
        if [service][type] {
      mutate { replace => { "[@metadata][archive][namespace]" => "none_name/%{+YYYY}/%{+MM}-%{+dd}/%{[service][type]}" } }
    } else {
      mutate { replace => { "[@metadata][archive][namespace]" => "none_name/%{+YYYY}/%{+MM}-%{+dd}/none_type" } }
    }
}

    if ![@metadata][archive][namespace] {
      mutate { replace => { "[@metadata][archive][namespace]" => "%{[@metadata][kafka][topic]}/%{+YYYY}/%{+MM}-%{+dd}" } }
    }
}

output {
   s3{
     access_key_id => "${S3_ACCESS_KEY_ID}"
     secret_access_key => "${S3_SECRET_ACCESS_KEY}"
     region => "ap-northeast-2"
     bucket => "${S3_BUCKET}"
     rotation_strategy => "size_and_time"
     size_file => 100000000
     time_file => 10
     codec => "json_lines"
     canned_acl => "bucket-owner-full-control"
     encoding => "gzip"
     prefix => "%{[@metadata][archive][namespace]}"
     temporary_directory => "/logstash/data/s3-temp"
     upload_workers_count => 16
     validate_credentials_on_root_bucket => false
     additional_settings => {
      force_path_style => true
      follow_redirects => false
     }
   }
}

 

 

 

logstash.yml

수정한 부분

#pipeline.workers: 1
pipeline.workers: 12
#
# How many workers should be used per output plugin instance
#
# pipeline.output.workers: 1
#
# How many events to retrieve from inputs before sending to filters+workers
#
# pipeline.batch.size: 125
pipeline.batch.size: 1024