この記事では、Yandex.Moneyが支払いサービスのサーバーログを収集および配信するシステムをどのように備えているかについて説明します。 私は昨年ログを配信してきましたが、この間に十分なニュアンスが蓄積され、一般の人と経験を共有することができました。

このシステムは、ほぼリアルタイムで動作することを目的として、EHKスタック（Elasticsearch / Heka / Kibana）に基づいて構築されています。 私は、1日に何十億行ものテキストを処理する繊細な場所とニュアンスに特に重点を置きます。

私は3つのことを信じています：監視、ログ、バックアップ。

Yandex.Moneyの運用ログにすばやくアクセスするために、 ElasticSearchが使用されます。ElasticSearchは 、支払いサービスを構成するほぼ100のマイクロサービスからデータを受け取ります。

「運用」とは、ログをリアルタイムで利用できることを意味します。ログファイルからElasticsearchクラスターに新しいエントリを配信する際の遅延は1秒未満になりました。 監視サービスはいつでもサービスの状態を常に正確に把握しており、開発者はリリース後すぐに各マイクロサービスの新しいバージョンの動作を迅速に評価および調整できます。

しかし、常にそうではありませんでした。 5か月前に会社に来たとき、 ElasticSearchクラスター（以降、単にES）への運用サービスログの配信に関する作業をセットアップするタスクが与えられました。 その時点で、分析とESへの配信の4つのスキームが使用されました。

• Heka-> TCP出力-> Heka-> ES。

  
  
  
  
  
  
  

• Heka→Apache Kafka→Apache Flume→ES。

  
  
  
  
  
  
  

• SyslogNg→ES。

  
  
  
  
  
  
  
• nxlog→nxlog→ES。

それらのほとんどすべてが不完全に機能しました。Kafkaクラスターは信頼性が低く、Flumeが定期的にハングし、ESが混乱に陥りました。

ログは実際には非常に単純なものです。バトルサーバーで急速にボリュームが増大する多くのファイルです。 したがって、この場合の単純なソリューションが最も信頼性が高くなります。

「傘の中のジョイントが多いほど、破損する可能性が高くなる」ため、私はすべてのアイドルで不当に複雑なサーキットをすべて捨て、ヘカがハンドラーと輸送の役割を果たすサーキットに落ち着きました。

Heka はログを取得し、TCPを介して別のHekaに送信して、ElasticSearchクラスターにさらに転送します。

その考えは次のとおりです。Hekaは各サーバーにインストールされ、このサーバーのすべてのサービスのログはProtobufバイナリプロトコルにパックされ、分析、フィルタリング、ESへの送信のためにHeka受信機に送信されます。

なぜ古典的なELKスタックではなく、LogstashではなくHekaを非推奨にするのか

ソフトウェアの構成と構成に関して、ESクラスターは次のとおりです。

• ElasticSearch 2.2。

  
  
  
  
  
  
  

• キバナ4.4.2。

  
  
  
  
  
  
  

• 2つのマスターノード：Intel Xeon 2x E5-2660、64 GB RAM、2x 146 GB RAID-10。

  
  
  
  
  
  
  

• Kibanaがインストールされたクライアントノード：Intel Xeon 2xE5-2660、64 GB RAM、2x146 GB RAID-10。

  
  
  
  
  
  
  
• 4つのデータノード：Intel Xeon 2x E5-2640 v3; 512 GB RAM、3x16 TB RAID-10。

すべてのESノードは同じローカルネットワーク上にあり、同じルーターに接続されているため、トランスポートプロトコルを使用してクラスター内で最大速度で通信できます。 このような組織は、インデックスの配置とクラスターの再バランスを著しく加速します。

現代の世界では、Elasticsearch / Logstash / Kibanaスタックは、ログを操作するための事実上の標準になっています。 ElasticsearchとKibanaに異議がない場合、Logstashには1つの警告があります。これはjRuby（Javaで書かれたRuby言語インタープリター）で作成され、システムにJVMが必要です。 Yandex.Moneyが数百の物理サーバーと仮想コンテナーに配置された多数のマイクロサービスであることを考えると、それぞれに重いLogstashとJAVAを配置するのは間違っています。 Hekaが選ばれた理由は、そのシンプルさ、信頼性、軽さ、通過メッセージのフィルタリング機能、および優れたデータバッファリングのためです。

製品のステータスに関しては（非推奨）-私たちにとってこれは議論ではありません。 軍事目的の弓矢も廃止されましたが、これはあなたが頭の中で誰かを撃つことを保証します。 たとえば、非標準のプラグインまたはプロセッサが必要な場合は、経験豊富な開発者のスタッフ全員が製品の完成を支援します。

しかし、それはすべて理論であり、実践への移行の問題が始まりました。

悪魔はデータ量に隠れていました

私たちの仕事の財務的な詳細を考えると、ほとんどすべてのサービスがログに多くの異なる情報を書き込みます。 スケールの例と理解：システムの一部のコンポーネントのログのボリュームは、 毎分25万行に達します。

ハードウェアがどれほど強力であっても、単一のHekaではなく、このようなボリュームだけを処理できます。パフォーマンスの低下とデータ損失は避けられません。 もちろん、どちらも完全に受け入れられるものではないため、 HAProxyが助けになります。 最終的なスキームは次のとおりです。

この図は、HekaソースからHAProxy + Hekaバンドルへのログトラフィックの一般的な方向を示しています。

各サーバーには、このサーバーのマイクロサービスのログを収集するHekaが1つあります。 データは収集され、Protobufにパッケージ化され、TCPを介してデータセンターにサービスを提供するロードバランサーに送信されます。 バックエンドはESデータノードに直接配置されたHAProxyであり、その背後にHekaプールがあります。 次に、データを受け入れ、ESJsonで再パッケージ化し、HTTPプロトコルを使用してローカルデータノードに送信します。

...さまざまなログファイル形式

会社の主な言語はJavaであり、ログは標準のlog4jライブラリを介して表示されるという事実にもかかわらず、「ドリームクラスター」の構築時に受け入れられた単一の形式はありませんでした。 各マイクロサービスは、表示フィールドのセットや日時形式のバリエーションなど、独自の種類のログを作成しました。 開発がログを一般的な形式に導くまで待ちたくなかったので、目標への動きは並行していました。 既存のログ形式の分析と同時に、修正のためにタスクが開始され、正しい形式の新しいバージョンがリリースされると、コレクターの設定が変更されました。

ジュース自体に移りましょう-設定のニュアンスです。

Hekaの動作は.tomlファイルによって記述され、起動時に単一の構成にアセンブルされ、記述されたブロックに応じて、Hekaはデータ処理モデルを構築します。

各ブロックにはいくつかの段階があります。

• 入力-入力ストリーム（ファイル、TCP / UDP入力、Kafkaから読み取ったデータ、Dockerコンテナーのイベントなど）。

  
  
  
  
  
  
  

• スプリッター-ここでは、ストリーム内の各データブロックの開始と終了を示します。 Java Stacktraceのような複数行データの必須ステップ。

  
  
  
  
  
  
  

• デコーダー-受信データをデコードするためのルールを説明します。 主に正規表現デコーダーを使用します。

  
  
  
  
  
  
  

• フィルター-データのフィルター処理と変更の段階。

  
  
  
  
  
  
  

• エンコーダー-受信者形式のデータストリームをエンコードします。

  
  
  
  
  
  
  
• 出力-ここでは、データがどのように、どこに行くべきかを説明します。

これらのすべての手順は、 GoおよびLuaのプラグインにすぎません。 必要に応じて、独自のものを作成できます。 たとえば、Luaのフィルタープラグインは、ESへの監視サービスリクエストの送信を遮断します。 または、ログから機密データを切り取ります。

古代エジプト神話のヒャカは魔法の神です。 そして、Hekaがログを使用してできることは単純に魔法です。

ログソースサーバーの設定

ログのソースサーバーとservice.tomlファイルの例を使用して、Hekaの構成を分析しましょう。

[money-service-log] type = "LogstreamerInput" log_directory = "/var/log/tomcat" file_match = "money-service.log" splitter = "RegexSplitter" decoder = "service_decoder"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

最も単純なケースは単一のファイルで、システムの手段によってローテーションが発生します。 多数のファイルがあり、形式が異なる場合は、入力/デコーダーの各ペアを記述する必要があります。 より詳細な説明については、公式ドキュメントを参照することをお勧めします 。 不明な点が残っている場合は、コメントで確認してください。

 [RegexSplitter] delimiter = '\n(\[\d\d\d\d-\d\d-\d\d)' delimiter_eol = false
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ログは複数行（同じスタックトレース）になる可能性があるため、Hekaがテキストのあるブロックが終了して別のブロックが開始する場所を理解できるRegexSplitterを忘れないでください。

 [service_decoder] type = "PayloadRegexDecoder" match_regex = '^\[(?P<timestamp>\d{4}-\d{2}-\d{2}T[\d:.\+]+\])\s+(?P<level>[AZ]+)\s+\[(?P<thread>.*)\]\s+\[(?P<context>\S*)\]\s+\[(?P<traceid>\S*)\]\s+\[(?P<unilabel>\S*)\]\s\[(?P<class>\S+)\]\s-\s(?P<msg>.*)' log_errors = true [service_decoder.message_fields] @timestamp = "%timestamp%" level = "%level%" thread = "%thread%" context = "%context%" traceid = "%traceid%" unilabel = "%unilabel%" class = "%class%" msg = "%msg%"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

match_regexでは 、Go言語標準の正規表現を使用してログ行を記述します。 Goの正規表現は、標準のPCREとほぼ一致していますが、Hekaの開始を拒否する可能性のあるニュアンスがいくつかあります。 たとえば、一部のPCRE実装では、次の構文を許可します。

 (?<groupname>.*)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

しかし、GOLANGは許しません。

log_errorsパラメーターを使用して、すべてのエラーを個別のログに収集します-後で必要になります。

 [ServiceOutput] type = "TcpOutput" address = "loadbalancer.server.address:port" keep_alive = true message_matcher = "Logger == 'money-appname'" use_buffering = true [ServiceOutput.buffering] max_file_size = 100857600 max_buffer_size = 1073741824 full_action = "block"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

流出バッファリングは、Hekaの優れた機能の1つです。 デフォルトでは、出力バッファは次のパスに保存されます。

 /var/cache/hekad/output_queue/OutputName
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

設定では、各バッファーファイルのサイズを100 MBに制限し、各出力モジュールの合計キャッシュサイズを1 GBに設定します。 full_actionパラメーターには、次の3つの値を指定できます。

• シャットダウン -バッファーがオーバーフローすると、Hekaは停止します。

  
  
  
  
  
  
  

• drop-バッファがオーバーフローすると、スタックのように動作を開始し、キュー内の古いメッセージを削除します。

  
  
  
  
  
  
  
• ブロック -バッファーがいっぱいになると、Hekaはすべての操作を一時停止し、データを送信できるようになるまで待機します。

ブロックを使用すると、ログの1行が失われないことが保証されます。 唯一のマイナスは、接続が切断されたりチャネルが低下した場合、接続が再開したときにトラフィックが急激に増加することです。 これは、Hekaが蓄積されたバッファーを送信し、リアルタイムで処理に戻ろうとするためです。 受信プールは余裕を持って設計する必要があります。そのような状況の可能性を必ず考慮してください。そうしないと、DDoS自体を簡単に変更できます。

ところで、Heka構成での二重引用符と単一引用符の使用に関しては、次のことが暗示されています。

• 単一引用符で囲まれた変数の値は、デフォルトでは生の文字列として扱われます。 特殊文字をスクリーニングする必要はありません。

  
  
  
  
  
  
  
• 二重引用符で囲まれた変数の値はプレーンテキストとして扱われ、変数で正規表現を使用する場合は二重エスケープが必要です。

やがてこのニュアンスは私に多くの血をだましました。

バックエンド構成

バランサーのバックエンドは、各ESデータノード上のHAProxyと3つのHekaインスタンスの束です。

HAProxyでは、すべてが非常に単純であり、説明を必要としないようです：

 listen pool_502 bind 0.0.0.0:502 balance roundrobin default-server fall 5 inter 5000 weight 10 server heka_1502 127.0.0.1:1502 check server heka_2502 127.0.0.1:2502 check server heka_3502 127.0.0.1:3502 check
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

各サーバーは、ポートのみが異なる3つのHekaインスタンスを実行します。

 [Input_502_1] type = "TcpInput" address = "0.0.0.0:1502" keep_alive = true keep_alive_period = 180 decoder = "MultiDecoder_502" [MultiDecoder_502] type = "MultiDecoder" subs = ['Service1Decoder', 'Service2Decoder'] cascade_strategy = "first-wins" log_sub_errors = true
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

多くのサービスのログが1つのポートを通過するため、構成ではMultiDecoderを使用します。 先勝ポリシーは、最初の一致の後、デコーダーのさらなる列挙が停止することを意味します。

 [Service1Decoder] type = "ProtobufDecoder" [Service2Decoder] type = "ProtobufDecoder" [Service1Encoder] type = "ESJsonEncoder" index = "service1-logs-%{%Y.%m.%d}" es_index_from_timestamp = false type_name = "%{Logger}" fields = [ "DynamicFields", "Payload", "Hostname" ] dynamic_fields = ["@timestamp", "level", "thread", "context", "traceid", "unilabel", "class", "msg"] [Service1Encoder.field_mappings] Payload = "message" Hostname = "MessageSourceAddress"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

es_index_from_timestampパラメーターは 、インデックス名を形成する日時が受信データからではなく、サーバーの現地時間から取得されることを示すために必要です。 サーバーが異なるタイムゾーンで動作し、誰かがUTCでタイムログをログに書き込み、誰かがMSKに書き込む場合の混乱を回避できます。

インデックスの原則は「1つのサービス-1つのインデックス」の原則を実装しており、毎日新しいインデックスが作成されます。

 [Service1Output] type = "ElasticSearchOutput" message_matcher = "Logger == 'money-service1'" server = "http://localhost:9200" encoder = "Service1Encoder" use_buffering = true
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

出力プラグインは、ログファイル名に対応するmessage_matcherパラメーターに基づいてデータストリームを解析します。 ネットワークを過負荷にしないために、Hekaはデータがインストールされているローカルデータノードにデータを送信します。 さらに、ES自体は、クラスターデータノード間のトランスポートプロトコルを介してインデックス付きデータを送信します。

おわりに

上記のスキームは正常に機能し、1秒あたり25〜3万レコードのインデックスを作成します。 Heka受信プールの安全マージンにより、最大10万レコード/秒の負荷ピークに耐えることができます。

Zabbixからの統計。

ElasticSearchでは、過去21日間のログを保存します。 経験から、古いデータへのアクセスはほとんど必要ないことが示されています。 ただし、必要に応じて、データをほぼ永久に保存するログサーバーからいつでも要求できます。

Kopfによるクラスターの現在の状態。

ログの収集と配信に関連するシステムの部分のみを説明したため、次の記事ではElasticSearchクラスター自体とその構成について説明します。 人類への信頼を失うことなく、仮想化の方法、バージョン2.2から5.3に移行し、240億件のレコードをどのように転送したかを伝えることを考えています。

ストーリーに何か他のものを追加したり、詳細を見失ったこともあるでしょうか？ コメントであなたの意見を共有してください。