How to Configure Logstash to Ship Logs to Alibaba Cloud OSS

Install Logstash (omitted)

Install logstash-output-oss plugin

/data/logstash/bin/logstash-plugin install logstash-output-oss

logstash.conf configuration

Input from Kafka:

input {
  kafka {
    bootstrap_servers => ["kafka1:9092,kafka2:9092,kafka3:9092"]
    group_id => 'oss_logs_group'
    topics => ["xinlimei_pro_topic","xinlimei_logs_pro","sunmei_logs_pro_topic"]
    client_id => "oss_logs_1"
    consumer_threads => 6
    sasl_mechanism => "PLAIN"
    security_protocol => "SASL_PLAINTEXT"
    sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='kafkasunmei0227';"
    codec => "json"
    auto_offset_reset => "latest"
  }
}

Filter section:

filter {
  mutate {
    lowercase => ["app_name"]
    remove_field => ["_index","_id","_type","_version","_score","referer","agent","@version","host"]
  }
  date {
    match => ["date", "yyyy-MM-dd HH:mm:ss.SSS"]
    target => '@timestamp'
    timezone => 'Asia/Shanghai'
  }
  ruby {
    code => "event.set('index_day', (event.get('@timestamp').time.localtime).strftime('%Y.%m.%d'))"
  }
}

Output to OSS:

output {
  oss {
    endpoint => "http://oss-cn-shanghai-internal.aliyuncs.com"
    bucket => "tr-app-logs"
    access_key_id => "your access_key"
    access_key_secret => "your access_key_secret"
    prefix => "%{index_day}-%{[app_name]}-%{[profiles_active]}"
    recover => true
    rotation_strategy => "size_and_time"
    time_rotate => 10
    size_rotate => 104857600
    temporary_directory => "/data/logstash/temporary/"
    encoding => "gzip"
    additional_oss_settings => {
      max_connections_to_oss => 1024
      secure_connection_enabled => false
    }
    codec => json {
      charset => "UTF-8"
    }
  }
}