
最近又在規劃ELK的設定，這次比較不一樣的地方我選擇了elastic cloud以及服務部署的方式都採用PaaS的方式作為部署，
加掛volume或是在機器上安裝filebeat都是一個比較困難的事情，所以一開始考慮使用azure blob queue的方式存放log，
但後來選擇了官方有提供的input套件，Azure Event Hub來寫log在使用logstash去讀取傳送到elasticsearch上

根據官網的文件操作後會一直出現
`The configuration will result in overwriting offsets. Please ensure that the each Event Hub's consumer_group is using a unique storage container.`
這樣的錯誤訊息，也採用了進階的設定去使用，但因為只有一組採用進階的方式設定有點太過複雜所以又改回原本設定，後來想我的event hub的connection string有兩組會不會因為這樣我需要多個storage container的存放空間，
後來將其中一組刪除後即可正常運作

```config
# 會出錯的logstash pipeline config
input {
    azure_event_hubs {
        event_hub_connections => ["Endpoint=sb://<<event hub>>.servicebus.windows.net/;SharedAccessKeyName=logsta...",
            "Endpoint=sb://<<event hub>>.servicebus.windows.net/;SharedAccessKeyName=logsta..."]
        storage_connection => "DefaultEndpointsProtocol=https;..."
        consumer_group => "logstash"
        decorate_events => true
        threads => 8
    }
}
filter {
    json {
        source => "message"
    }
    date {
        match => [ "Timestamp", "ISO8601" ]
        target => "@timestamp"
    }
    mutate {
        rename => ["MessageTemplate", "message" ]
        rename => ["Level", "level" ]
        merge => { "message" => "Exception" }
        remove_field => ["Exception", "Timestamp"]
    }
}
output {
    elasticsearch {
        cloud_id => "<<Cloud id>>"
        cloud_auth => "<<user>>:<<password>>"
        index => "demo-%{+YYYY.w}"
    }
}
```

```config
# 最後的 logstash pipeline config
input {
    azure_event_hubs {
        event_hub_connections => ["Endpoint=sb://<<event hub>>.servicebus.windows.net/;SharedAccessKeyName=logsta..."]
        storage_connection => "DefaultEndpointsProtocol=https;..."
        consumer_group => "logstash"
        decorate_events => true
        threads => 8
    }
}
filter {
    json {
        source => "message"
    }
    date {
        match => [ "Timestamp", "ISO8601" ]
        target => "@timestamp"
    }
    mutate {
        rename => ["MessageTemplate", "message" ]
        rename => ["Level", "level" ]
        merge => { "message" => "Exception" }
        remove_field => ["Exception", "Timestamp"]
    }
}
output {
    elasticsearch {
        cloud_id => "<<Cloud id>>"
        cloud_auth => "<<user>>:<<password>>"
        index => "demo-%{+YYYY.w}"
    }
}
```


### 參考資料
[Github - logstash-input-azure_event_hubs](https://github.com/logstash-plugins/logstash-input-azure_event_hubs)


最近又在規劃ELK的設定，這次比較不一樣的地方我選擇了elastic cloud以及服務部署的方式都採用PaaS的方式作為部署， 加掛volume或是在機器上安裝filebeat都是一個比較困難的事情，所以一開始考慮使用azure blob queue的方式存放log， 但後來選擇了官方有提供的input套件，Azure Event Hub來寫log在使用logstash去讀取傳送到elasticsearch上

Logstash Azure event hub input 設定

參考資料

文章標籤